这是一篇测试文章

博客园 - Artech

这是一篇测试文章

Artech · 2026-05-18 · via 博客园 - Artech

FastMCP是对MCP规范的实现，其消息内容统一采用JSON-RPC 2.0格式。在底层传输层面，FastMCP主要支持In-Memory，STDIO 、Streamable-HTTP和SSE协议。

1. ClientTransport

FastMCP的传输层旨在管理客户端和MCP服务器之间的底层连接。FastMCP主要支持STDIO 、Streamable-HTTP和SSE三种协议。从客户端角度来讲，虽然它可以根据您传递的信息自动解析出传输类型（比如根据指定URL的路径模式确定采用SSE还是Streamable-HTTP），但显式指定代表传输的ClientTransport对象可以让我们对传输具有完全的掌控。

class ClientTransport(abc.ABC):
    @abc.abstractmethod
    @contextlib.asynccontextmanager
    async def connect_session(
        self, **session_kwargs: Unpack[SessionKwargs]
    ) -> AsyncIterator[ClientSession]
    async def close(self)
    def get_session_id(self) -> str | None
    def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None)

ClientTransport定义了客户端如何与服务器建立连接、交换数据以及管理连接生命周期的标准“合同”，它定义了如下几个核心方法：

connect_session：负责建立物理连接（如打开STDIO管道或发起HTTP请求），并将其升级为一个逻辑会话。由于被装饰为@contextlib.asynccontextmanager（异步上下文管理器），我们一般采用async with模式调用此方法；
close：提供了一个标准的资源清理接口。当不再需要连接时（例如程序退出），调用此方法来关闭文件描述符、停止子进程或关闭HTTP客户端连接池；
get_session_id：主要用于HTTP/SSE传输模式下获取Session ID。由于HTTP是无状态的，需要一个ID来标识当前这个逻辑连接；
-_set_auth：如果采用HTTP传输的服务端提供了的认证，客户端可以重写此方法来处理API Key或OAuth令牌。对于STDIO传输模式不需要它；

如果调用FastMCP的run方法时，没有利用stateless_http参数将其设置成无状态的服务器，客户端和服务端之间的交互都会一个Session中进行。Sessions是客户端与服务端之间通信的核心生命周期单位。每个Session代表一个完整的交互过程，该过程包含三个主要阶段：

初始化：客户端连接到服务端后，双方交换Initialize请求。这包括能力协商，即双方将各自支持的能力和特性提交给对方，后续会在处理请求的时候会成分考虑对方的能力范围；
交互：在Session存续期间，客户端可以按需调用服务端提供的工具或请求资源数据，服务端也可以反向发送请求和通知；
终止：会话关闭时，资源会被释放。如果是通过STDIO传输，进程退出即代表会话结束；如果是SSE通常由客户端主动断开连接；

ClientTransport的connect_session方法返回代表客户端会话的ClientSession对象，我们说客户端和服务端之间的交互在一个确定的会话中进行，具体体现在ClientSession提供了几乎所有与服务端进行交互的方法。Client用于操作工具、资源和提示词的方法最终都会转发到ClientSession对象上，它的session特性返回此对象。ClientSession类型由mcp库提供，mcp库是对MCP协议的官方实现。

class Client(
    Generic[ClientTransportT],
    ClientResourcesMixin,
    ClientPromptsMixin,
    ClientToolsMixin,
    ClientTaskManagementMixin,
):
    @property
    def session(self) -> ClientSession

2. In-Memory

表示FastMCP客户端的Client对象可以直接根据FastMCP对象来创建。这种方式相当于让服务器和客户端共享同一进程，客户端的调用直接转发给FastMCP对象，这无疑使最高效的通信方式。我们称这种方式为In-Memmory传输，本质它们就是共享同一进程内的内存空间进行通信。这种传输形式在FastMCP中通过如下这个FastMCPTransport类型表示。

class FastMCPTransport(ClientTransport):
    def __init__(self, mcp: FastMCP | FastMCP1Server, raise_exceptions: bool = False):
    @contextlib.asynccontextmanager
    async def connect_session(
        self, **session_kwargs: Unpack[SessionKwargs]
    ) -> AsyncIterator[ClientSession]:

我们通过如下这个实例来验证这种传输方式下服务器和客户端共享进程。我们在创建的FastMCP对象中注册了一个用于返回当前进程ID的工具函数get_process_id，并利用此FastMCP对象创建了一个Client对象。我们通过工具调用得到服务器进程ID，并通过断言验证它与当前客户端进程ID一致。

from fastmcp import FastMCP, Client
import asyncio
import os

mcp = FastMCP()
@mcp.tool()
async def get_process_id() -> int:
    """Get the current process ID"""
    return os.getpid()

client = Client(mcp)

async def main():    
    async with client:
        result = await client.call_tool(name="get_process_id", arguments= {})
        server_process_id = int(result.content[0].text) if result.content else -1 # type: ignore
        client_process_id = os.getpid()
        assert server_process_id ==client_process_id

if __name__ == "__main__":   
    asyncio.run(main())

3. STDIO

STDIO基于标准输入/输出，是FastMCP的默认传输方式，专为本地开发和桌面应用设计。在这种传输模式下，客户端将服务器作为一个子进程启动，通过标准输入（stdin）和标准输出（stdout）发送请求和接收响应。由于采用跨进程通信，其极低延迟，无需配置网络端口或身份验证，安全性高,所以适合本地工具集成、CLI工具开发。 STDIO传输通过StdioTransport类型表示。

class StdioTransport(ClientTransport):
    def __init__(
        self,
        command: str,
        args: list[str],
        env: dict[str, str] | None = None,
        cwd: str | None = None,
        keep_alive: bool | None = None,
        log_file: Path | TextIO | None = None,
    )

上面给出了StdioTransport构造函数的定义，它具有如下的参数：

command：用于启动MCP服务器的命令；
args：为MCP服务器启动命令提供的参数列表；
env：为MCP服务器进程设置的环境变量；
cwd：启动MCP服务器进程采用的当前工作目录；
keep_alive:决定当连接异常或空闲时，是否尝试维持或重启进程；
log_file:日志重定向目标。由于STDIO使用stdout传输数据，所以我们绝对不能在服务端代码里直接print函数调试信息，否则会破坏协议格式导致崩溃。

我们编写了如下这个程序来演示keep_alive参数针对服务器进程的重用。如下所示的是作为MCP服务器的脚本（mcp-server.py），其中定义了一个用于返回服务进程ID的工具函数get_process_id。

from fastmcp import FastMCP
import os
mcp = FastMCP()
@mcp.tool()
async def get_process_id() -> int:
    """Get the current process ID"""
    return os.getpid()
mcp.run()

在如下的客户端程序中，main函数会利用自身的参数keep_alive去创建对应的StdioTransport。当Client对象根据这个StdioTransport创建出来后，我们在两个会话中调用工具get_process_id，并输出作为返回值的服务器进程ID。

from fastmcp import Client
from fastmcp.client.transports import StdioTransport
from pathlib import Path
import asyncio

async def main(keep_alive:bool):
    transport = StdioTransport(
    command="python",
    args=["mcp-server.py"],
    cwd=str(Path(__file__).parent),
    keep_alive= keep_alive)

    client = Client(transport)
    async with client:
        result = await client.call_tool(name="get_process_id", arguments= {})
        print(f"Server process ID: {result.content[0].text}") # type: ignore      

    async with client:
        result = await client.call_tool(name="get_process_id", arguments= {})
        print(f"Server process ID: {result.content[0].text}\n") # type: ignore

if __name__ == "__main__":   
    asyncio.run(main(keep_alive=True))    
    asyncio.run(main(keep_alive=False))

从如下的输出可以看出，如果创建StdioTransport时将keep_alive参数设置为True，Session结束之后，服务器进程并不会关闭，并会被后续Session复用。

Server process ID: 6264
Server process ID: 6264
Server process ID: 14588
Server process ID: 31308

4. SSE(Server Send Event)

SSE是基于HTTP的单向推送协议，允许服务器通过一条持久的HTTP连接持续向客户端推送数据。SSE是单向的（只能从服务器推送到客户端）,为了实现客户端与MCP服务端之间的“双向对话”，它采用了“双通道”设计：

下行通道 (SSE Connection): 这是一个从服务端到客户端的长连接。客户端请求服务器的一个特定端点（路径为/sse），服务器保持连接不挂断。服务器通过这个通道把工具执行结果、通知、进度等“推”给客户端;
上行通道 (POST Request)：这是一个从客户端到服务端的短连接（路径为/messages）。每当客户端想要调用一个工具或发送指令时，它会发起一个标准的POST请求，发完连接就断开了;

服务器怎么知道POST请求里的指令该把结果推给哪个SSE连接呢？这需要借助于Session ID对客户端的标识作用。当客户端第一次建立SSE连接时，服务器会通过这个连接发回一个唯一的Session ID。客户端后续的所有POST请求都会带上这个ID。服务器收到POST后，查一下ID，就知道该把处理结果塞进某个SSE长连接里发回去了。为什么不像WebSocket那样直接用一个连接呢？主要由如下的原因：

防火墙友好：很多公司内网防火墙会拦截WebSocket，但很少拦截普通的HTTP POST和长轮询；
Web 标准：SSE是原生的Web标准，不需要复杂的握手过程，实现起来比WebSocket轻量得多；
无状态性：上行通道（POST）是无状态的，方便负载均衡；只有下行通道（SSE）需要维护简单的连接状态；

基于SSE的传输在FastMCP中通过SSETransport类型表示。

class SSETransport(ClientTransport):
    def __init__(
        self,
        url: str | AnyUrl,
        headers: dict[str, str] | None = None,
        auth: httpx.Auth | Literal["oauth"] | str | None = None,
        sse_read_timeout: datetime.timedelta | float | int | None = None,
        httpx_client_factory: McpHttpClientFactory | None = None,
        verify: ssl.SSLContext | bool | str | None = None,
    )

上面的代码给出了SSETransport构造函数的定义，具体的参数如下：

url：SSE服务的入口地址。如果FastMCP服务器以SSE传输方式启动，服务端到客户端连接对应的路径会设置为/sse，此参数指向的正是这个路径；
headers：默认添加的请求报头；
auth：身份验证配置。支持多种认证方式，可以是简单的 (user, password) 元组，也可以是复杂的OAuth流程；
sse_read_timeout：由于SSE是长连接，如果服务器长时间（此参数设定）不发数据，连接可能会被中间代理断开；
httpx_client_factory：HTTP客户端工厂函数，可以利用它注入一些钩子参与HTTP请求和响应的处理；
verify：SSL/TLS证书校验；

FastMCP服务器启动的时候，如果希望采用SSE传输协议，可以按照如下的方式在调用run方法时将transport参数设置为sse。

from fastmcp import FastMCP
mcp = FastMCP("Greeting")

@mcp.tool()
async def greet(name: str) -> str:
    """Get a greeting message for the given name"""
    return f"Hi, {name}!"

mcp.run(transport="sse", host="0.0.0.0", port=3721)

对应的客户端程序如下所示，用于创建Client的SSETransport需要将URL设置为http://localhost:3721/sse。

import asyncio
from fastmcp import Client
from fastmcp.client.transports import SSETransport

client = Client(SSETransport(url="http://localhost:3721/sse")) 

async def main():
    async with client:
        await client.call_tool(name="greet", arguments= {"name": "Alice"})
        await client.call_tool(name="greet", arguments= {"name": "Bob"})
if __name__ == "__main__":
    asyncio.run(main())

上面这个程序会涉及若干HTTP往复，其中第一次HTTP消息交换是为了建立SSE通道，具体请求和回复内容如下:

请求：

GET http://localhost:3721/sse HTTP/1.1
Host: localhost:3721
Accept-Encoding: gzip, deflate, zstd
Connection: keep-alive
User-Agent: python-httpx/0.28.1
Accept: text/event-stream
Cache-Control: no-store

响应：

HTTP/1.1 200 OK
date: Sun, 29 Mar 2026 08:35:11 GMT
server: uvicorn
cache-control: no-store
connection: keep-alive
x-accel-buffering: no
content-type: text/event-stream; charset=utf-8
Transfer-Encoding: chunked

可以看出SSE通道对应终结点采用的路径为/sse。如下所示的是工具调用的请求和回复，可以看出客户端请求通道对应终结点的路径为/messages。每次请求会利用查询字符串携带Session ID，由于响应内容是通过SSE通道推送给客户端的，所以得到的仅仅是一个202 Accepted响应。

请求：

POST http://localhost:3721/messages/?session_id=d7e8cee1e85a4bdfbd7dcfda2f25a7c9 HTTP/1.1
Host: localhost:3721
Accept: */*
Accept-Encoding: gzip, deflate, zstd
Connection: keep-alive
User-Agent: python-httpx/0.28.1
Content-Length: 127
Content-Type: application/json

{"method":"tools/call","params":{"name":"greet","arguments":{"name":"Bob"},"_meta":{"progressToken":3}},"jsonrpc":"2.0","id":3}

响应：

HTTP/1.1 202 Accepted
date: Sun, 29 Mar 2026 08:35:13 GMT
server: uvicorn
content-length: 8

Accepted

5. Streamable-HTTP

SSE已经是一个过时的协议，Streamable-HTTP为SSE的升级版。和SSE一样，Streamable-HTTP也通过建立两个连接的方式实现双工通信，但它实现得更加灵活：

两个连接对应的终结点共享相同的路径/mcp，而SSE的两个通道对应的终结点的路径分别为/sse和/messages；
客户端利用上行通道发送POST请求执行相应的操作，如果操作没用采用后台任务的形式被调度执行，会立即执行返回的结果会利用此连接返回；SSE总是利用下行通道（sse长连接）以通知的形式返回操作执行的结果；
由于上行通道可以用于响应POST请求的结果，下行通道未必能用得上（比如在一个Session中就单纯地执行一次工具调用），所以SSE长连接会采用延迟创建的方式；SSE发送的第一个GET请求就是为了创建sse长连接；
如何支持HTTP2和HTTP3（QUIC），可以直接利用它们提供的多路复用，此时不必创建双连接；SSE会忽略通信双方针对HTTP2/3的支持；

基于Streamable-HTTP的传输通过StreamableHttpTransport类型表示，它和SSETransport的构造函数具有完全一致的参数列表。

class StreamableHttpTransport(ClientTransport):
    def __init__(
        self,
        url: str | AnyUrl,
        headers: dict[str, str] | None = None,
        auth: httpx.Auth | Literal["oauth"] | str | None = None,
        sse_read_timeout: datetime.timedelta | float | int | None = None,
        httpx_client_factory: McpHttpClientFactory | None = None,
        verify: ssl.SSLContext | bool | str | None = None,
    )

如果FastMCP服务希望采用Streamable-HTTP，可以采用如下的方式调用run方法的时候将transport参数设置为streamable-http。如果Client并非通过StreamableHttpTransport对象进行创建，而是直接指定一个URL，如果此地址包含/sse分段，则使用SSE，否则使用Streamable-HTTP。

from fastmcp import FastMCP
mcp = FastMCP("Greeting")

@mcp.tool()
async def greet(name: str) -> str:
    """Get a greeting message for the given name"""
    return f"Hi, {name}!"

mcp.run(transport="streamable-http", host="0.0.0.0", port=3721)

我们针对上面定义的这个FastMCP服务器定义了如下所示的客户端程序。我们利用提供的服务器地址创建了StreamableHttpTransport对象，并利用后者创建了一个Client对象。在利用Client对象创建的同一个Session中，我们调用了工具greet。

import asyncio
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport

client = Client(StreamableHttpTransport(url="http://localhost:3721/mcp")) 

async def main():
    async with client:
        await client.call_tool(name="greet", arguments= {"name": "MCP"})
        await asyncio.sleep(10) 
if __name__ == "__main__":
    asyncio.run(main())

和SSE不同，调用greet工具的结果可以直接在请求的响应中返回，而不是得到一个202 Accepted响应。具体的请求和响应如下所示，我们还会发现Session ID会通过请求的报头进行传递。

请求：

POST http://localhost:3721/mcp HTTP/1.1
Host: localhost:3721
Accept-Encoding: gzip, deflate, zstd
Connection: keep-alive
User-Agent: python-httpx/0.28.1
accept: application/json, text/event-stream
content-type: application/json
mcp-session-id: 9b8892b1a5ff4b18842569358c01d903
mcp-protocol-version: 2025-11-25
Content-Length: 129

{"method":"tools/call","params":{"name":"greet","arguments":{"name":"MCP"},"_meta":{"progressToken":1}},"jsonrpc":"2.0","id":1}

响应：

HTTP/1.1 200 OK
date: Sun, 29 Mar 2026 13:29:45 GMT
server: uvicorn
cache-control: no-cache, no-transform
connection: keep-alive
content-type: text/event-stream
mcp-session-id: 9b8892b1a5ff4b18842569358c01d903
x-accel-buffering: no
Content-Length: 169

event: message
data: {"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"Hi, MCP!"}],"structuredContent":{"result":"Hi, MCP!"},"isError":false}}

由于sse连接时延迟创建的，所以我们在完成工具调用后延迟了10秒钟才关闭Session，此时我们可以拦截到用于创建sse连接的请求和响应(如下所示)。如果没有这一个等待，创建sse连接的GET请求会在Session关闭之后被发送，此时就会得到一个404 Not Found响应。

请求：

GET http://localhost:3721/mcp HTTP/1.1
Host: localhost:3721
Accept-Encoding: gzip, deflate, zstd
Connection: keep-alive
User-Agent: python-httpx/0.28.1
accept: application/json, text/event-stream
content-type: application/json
mcp-session-id: 9b8892b1a5ff4b18842569358c01d903
mcp-protocol-version: 2025-11-25
Accept: text/event-stream
Cache-Control: no-store

响应：

HTTP/1.1 200 OK
date: Sun, 29 Mar 2026 13:29:47 GMT
server: uvicorn
cache-control: no-cache, no-transform
connection: keep-alive
content-type: text/event-stream
mcp-session-id: 9b8892b1a5ff4b18842569358c01d903
x-accel-buffering: no
Content-Length: 0

6. 多服务器客户端

一个Client可以同时连接多个MCP服务器，我们可以针对不同传输协议的MCP服务器可以定义在配置字典中。以如下这个Client为例，它连接了两个MCP服务器，一个采用HTTP传输协议（Streamable-HTTP），另一个则采用STDIO传输。

from fastmcp import Client

config = {
    "mcpServers": {
        "weather": {
            "url": "https://weather.example.com/mcp",
            "transport": "http"
        },
        "assistant": {
            "command": "python",
            "args": ["./assistant.py"],
            "env": {"LOG_LEVEL": "INFO"}
        }
    }
}

client = Client(config)

async with client:
    weather = await client.call_tool("weather_get_forecast", {"city": "NYC"})
    answer = await client.call_tool("assistant_ask", {"question": "What?"})

为了解决多MCP服务器之间的组件命名冲突，配置字典的Key会作为命名空间，所以上面调用的两个工具的名称前面会分别添加weather_和assistant_前缀。

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

博客园 - Artech

1. ClientTransport

2. In-Memory

3. STDIO

4. SSE(Server Send Event)

5. Streamable-HTTP

6. 多服务器客户端