FastMCP是对MCP规范的实现其消息内容统一采用JSON-RPC 2.0格式。在底层传输层面FastMCP主要支持In-MemorySTDIO、Streamable-HTTP和SSE协议。1. ClientTransportFastMCP的传输层旨在管理客户端和MCP服务器之间的底层连接。FastMCP主要支持STDIO、Streamable-HTTP和SSE三种协议。从客户端角度来讲虽然它可以根据传递的信息自动解析出传输类型比如根据指定URL的路径模式确定采用SSE还是Streamable-HTTP但显式指定代表传输的ClientTransport对象可以让我们对传输具有完全的掌控。classClientTransport(abc.ABC):abc.abstractmethodcontextlib.asynccontextmanagerasyncdefconnect_session(self,**session_kwargs:Unpack[SessionKwargs])-AsyncIterator[ClientSession]asyncdefclose(self)defget_session_id(self)-str|Nonedef_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中进行。Session是客户端与服务端之间通信的核心生命周期单位。每个Session代表一个完整的交互过程该过程包含三个主要阶段初始化客户端连接到服务端后双方交换Initialize请求。这包括能力协商即双方将各自支持的能力和特性提交给对方后续会在处理请求的时候会充分考虑对方的能力范围交互在Session存续期间客户端可以按需调用服务端提供的工具或请求资源数据服务端也可以反向发送请求和通知终止会话关闭时资源会被释放。如果是通过STDIO传输进程退出即代表会话结束如果是SSE通常由客户端主动断开连接。ClientTransport的connect_session方法返回代表客户端会话的ClientSession对象。我们说客户端和服务端之间的交互在一个确定的会话中进行也体现在ClientSession提供了几乎所有与服务端进行交互的方法。Client用于操作工具、资源和提示词的方法最终都会转发到ClientSession对象上它的session属性返回此对象。ClientSession类型由mcp库提供mcp库是对MCP协议的官方实现。classClient(Generic[ClientTransportT],ClientResourcesMixin,ClientPromptsMixin,ClientToolsMixin,ClientTaskManagementMixin,):propertydefsession(self)-ClientSession2. In-Memory表示FastMCP客户端的Client对象可以直接根据FastMCP对象来创建。这种方式相当于让服务器和客户端共享同一进程客户端的调用直接转发给FastMCP对象这无疑使最高效的通信方式。我们称这种方式为In-Memmory传输本质它们就是共享同一进程内的内存空间进行通信。这种传输形式在FastMCP中通过如下这个FastMCPTransport类型表示。classFastMCPTransport(ClientTransport):def__init__(self,mcp:FastMCP|FastMCP1Server,raise_exceptions:boolFalse):contextlib.asynccontextmanagerasyncdefconnect_session(self,**session_kwargs:Unpack[SessionKwargs])-AsyncIterator[ClientSession]:我们通过如下这个实例来验证这种传输方式下服务器和客户端共享进程。我们在创建的FastMCP对象中注册了一个用于返回当前进程ID的工具函数get_process_id并利用此FastMCP对象创建了一个Client对象。我们通过工具调用得到服务器进程ID并通过断言验证它与当前客户端进程ID一致。fromfastmcpimportFastMCP,Clientimportasyncioimportos mcpFastMCP()mcp.tool()asyncdefget_process_id()-int:Get the current process IDreturnos.getpid()clientClient(mcp)asyncdefmain():asyncwithclient:resultawaitclient.call_tool(nameget_process_id,arguments{})server_process_idint(result.content[0].text)ifresult.contentelse-1# type: ignoreclient_process_idos.getpid()assertserver_process_idclient_process_id asyncio.run(main())3. STDIOSTDIO基于标准输入/输出是FastMCP的默认传输方式专为本地开发和桌面应用设计。在这种传输模式下客户端将服务器作为一个子进程启动通过标准输入stdin和标准输出stdout发送请求和接收响应。由于采用跨进程通信其极低延迟无需配置网络端口或身份验证安全性高,所以适合本地工具集成、CLI工具开发。 STDIO传输通过StdioTransport类型表示。classStdioTransport(ClientTransport):def__init__(self,command:str,args:list[str],env:dict[str,str]|NoneNone,cwd:str|NoneNone,keep_alive:bool|NoneNone,log_file:Path|TextIO|NoneNone,)上面给出了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。fromfastmcpimportFastMCPimportos mcpFastMCP()mcp.tool()asyncdefget_process_id()-int:Get the current process IDreturnos.getpid()mcp.run()在如下的客户端程序中main函数会利用自身的参数keep_alive去创建对应的StdioTransport。当Client对象根据这个StdioTransport创建出来后我们在两个会话中调用工具get_process_id并输出作为返回值的服务器进程ID。fromfastmcpimportClientfromfastmcp.client.transportsimportStdioTransportfrompathlibimportPathimportasyncioasyncdefmain(keep_alive:bool):transportStdioTransport(commandpython,args[mcp-server.py],cwdstr(Path(__file__).parent),keep_alivekeep_alive)clientClient(transport)asyncwithclient:resultawaitclient.call_tool(nameget_process_id,arguments{})print(fServer process ID:{result.content[0].text})# type: ignoreasyncwithclient:resultawaitclient.call_tool(nameget_process_id,arguments{})print(fServer process ID:{result.content[0].text}\n)# type: ignoreasyncio.run(main(keep_aliveTrue))asyncio.run(main(keep_aliveFalse))输出Server process ID: 6264 Server process ID: 6264 Server process ID: 14588 Server process ID: 31308从如下的输出可以看出如果创建StdioTransport时将keep_alive参数设置为TrueSession结束之后服务器进程并不会关闭并会被后续Session复用。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类型表示。classSSETransport(ClientTransport):def__init__(self,url:str|AnyUrl,headers:dict[str,str]|NoneNone,auth:httpx.Auth|Literal[oauth]|str|NoneNone,sse_read_timeout:datetime.timedelta|float|int|NoneNone,httpx_client_factory:McpHttpClientFactory|NoneNone,verify:ssl.SSLContext|bool|str|NoneNone,)上面的代码给出了SSETransport构造函数的定义具体的参数如下urlSSE服务的入口地址。如果FastMCP服务器以SSE传输方式启动服务端到客户端连接对应的路径会设置为/sse此参数指向的正是这个路径headers默认添加的请求报头auth身份验证配置。支持多种认证方式可以是简单的 (user, password) 元组也可以是复杂的OAuth流程sse_read_timeout由于SSE是长连接如果服务器长时间此参数设定不发数据连接可能会被中间代理断开httpx_client_factoryHTTP客户端工厂函数可以利用它注入一些钩子参与HTTP请求和响应的处理verifySSL/TLS证书校验。FastMCP服务器启动的时候如果希望采用SSE传输协议可以按照如下的方式在调用run方法时将transport参数设置为sse。fromfastmcpimportFastMCP mcpFastMCP(Greeting)mcp.tool()asyncdefgreet(name:str)-str:Get a greeting message for the given namereturnfHi,{name}!mcp.run(transportsse,host0.0.0.0,port3721)对应的客户端程序如下所示用于创建Client的SSETransport需要将URL设置为http://localhost:3721/sse。importasynciofromfastmcpimportClientfromfastmcp.client.transportsimportSSETransport clientClient(SSETransport(urlhttp://localhost:3721/sse))asyncdefmain():asyncwithclient:awaitclient.call_tool(namegreet,arguments{name:Alice})awaitclient.call_tool(namegreet,arguments{name:Bob})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; charsetutf-8 Transfer-Encoding: chunked可以看出SSE通道对应终结点采用的路径为/sse。如下所示的是工具调用的请求和回复可以看出客户端请求通道对应终结点的路径为/messages。每次请求会利用查询字符串携带session_id由于响应内容是通过SSE通道推送给客户端的所以得到的仅仅是一个202 Accepted响应。请求POST http://localhost:3721/messages/?session_idd7e8cee1e85a4bdfbd7dcfda2f25a7c9 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 Accepted5. Streamable-HTTP对于FastMCP来说SSE已经是一个过时的协议Streamable-HTTP为SSE的升级版。和SSE一样Streamable-HTTP也通过建立两个连接的方式实现双工通信但它实现得更加灵活两个连接对应的终结点共享相同的路径/mcp而SSE的两个通道对应的终结点的路径分别为/sse和/messages客户端利用上行通道发送POST请求执行相应的操作如果操作没用采用后台任务的形式被调度执行会立即执行返回的结果会利用此连接返回SSE总是利用下行通道sse长连接以通知的形式返回操作执行的结果由于上行通道可以用于响应POST请求的结果下行通道未必能用得上比如在一个Session中就单纯地执行一次工具调用所以SSE长连接会采用延迟创建的方式SSE发送的第一个GET请求就是为了创建sse长连接如何支持HTTP2和HTTP3QUIC可以直接利用它们提供的多路复用此时不必创建双连接SSE会忽略通信双方针对HTTP2/3的支持基于Streamable-HTTP的传输通过StreamableHttpTransport类型表示它和SSETransport的构造函数具有完全一致的参数列表。classStreamableHttpTransport(ClientTransport):def__init__(self,url:str|AnyUrl,headers:dict[str,str]|NoneNone,auth:httpx.Auth|Literal[oauth]|str|NoneNone,sse_read_timeout:datetime.timedelta|float|int|NoneNone,httpx_client_factory:McpHttpClientFactory|NoneNone,verify:ssl.SSLContext|bool|str|NoneNone,)如果FastMCP服务希望采用Streamable-HTTP可以采用如下的方式调用run方法的时候将transport参数设置为streamable-http。如果Client并非通过StreamableHttpTransport对象进行创建而是直接指定一个URL如果此地址包含/sse分段则使用SSE否则使用Streamable-HTTP。fromfastmcpimportFastMCP mcpFastMCP(Greeting)mcp.tool()asyncdefgreet(name:str)-str:Get a greeting message for the given namereturnfHi,{name}!mcp.run(transportstreamable-http,host0.0.0.0,port3721)我们针对上面定义的这个FastMCP服务器定义了如下所示的客户端程序。我们利用提供的服务器地址创建了StreamableHttpTransport对象并利用后者创建了一个Client对象。在利用Client对象创建的同一个Session中我们调用了工具greet。importasynciofromfastmcpimportClientfromfastmcp.client.transportsimportStreamableHttpTransport clientClient(StreamableHttpTransport(urlhttp://localhost:3721/mcp))asyncdefmain():asyncwithclient:awaitclient.call_tool(namegreet,arguments{name:MCP})awaitasyncio.sleep(10)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: 06. 多服务器客户端一个Client可以同时连接多个MCP服务器我们可以将针对不同传输协议的MCP服务器定义在配置字典中。以如下这个Client为例它连接了两个MCP服务器一个采用HTTP传输协议Streamable-HTTP另一个则采用STDIO传输。fromfastmcpimportClient config{mcpServers:{weather:{url:https://weather.example.com/mcp,transport:http},assistant:{command:python,args:[./assistant.py],env:{LOG_LEVEL:INFO}}}}clientClient(config)asyncwithclient:weatherawaitclient.call_tool(weather_get_forecast,{city:NYC})answerawaitclient.call_tool(assistant_ask,{question:What?})为了解决多MCP服务器之间的组件命名冲突配置字典的Key会作为命名空间所以上面调用的两个工具的名称前面会分别添加weather_和assistant_前缀。