[FastMCP设计、原理与应用-04]正常交流之前,先了解对方的能力!

张开发
2026/4/13 17:51:35 15 分钟阅读

分享文章

[FastMCP设计、原理与应用-04]正常交流之前,先了解对方的能力!
MCP协议针对服务器和客户端定义了很多特性它们并非在所有的MCP服务器和客户端中都会被支持所以客户端和服务器在交互的初始阶段需要交换彼此支持的特性集这是接下来得以正常工作的基础。1. 服务端能力我们将针对特性的支持与否视为MCP服务器和客户端的能力服务器能力通过如下这个ServerCapabilities类型表示它由mcp库提供模块路径为mcp.types。它利用定义了关于工具、资源、提示词、日志、后台任务、自动补全以及一些实验性的能力。这是一个Pydantic模型model_config字段的设置ConfigDict(extra“allow”)还赋予它自由扩展的能力。classServerCapabilities(BaseModel):experimental:dict[str,dict[str,Any]]|NoneNonelogging:LoggingCapability|NoneNoneprompts:PromptsCapability|NoneNoneresources:ResourcesCapability|NoneNonetools:ToolsCapability|NoneNonecompletions:CompletionsCapability|NoneNone tasks:ServerTasksCapability|NoneNonemodel_configConfigDict(extraallow)classPromptsCapability(BaseModel):listChanged:bool|NoneNonemodel_configConfigDict(extraallow)classResourcesCapability(BaseModel):subscribe:bool|NoneNonelistChanged:bool|NoneNonemodel_configConfigDict(extraallow)classToolsCapability(BaseModel):listChanged:bool|NoneNonemodel_configConfigDict(extraallow)classLoggingCapability(BaseModel):model_configConfigDict(extraallow)classCompletionsCapability(BaseModel):model_configConfigDict(extraallow)classServerTasksCapability(BaseModel):model_configConfigDict(extraallow)list:TasksListCapability|NoneNonecancel:TasksCancelCapability|NoneNonerequests:ServerTasksRequestsCapability|NoneNone可供配置的能力集其实不多其中ToolsCapability、ResourcesCapability和PromptsCapability的listChanged字段表示MCP服务器提供的工具、资源和提示词列表发生改变后是否具有向客户端发送通知的能力。其余能力包括ResourcesCapabilitysubscribe服务器是否支持客户端针对某个资源的“订阅”。如果设置为True客户端可以向服务端发送一个resources/subscribe请求订阅某个资源。一旦订阅成功当该资源的内容发生变化时服务端会自动发消息通知客户端而不需要客户端每隔一段时间去轮询一次ServerTasksCapability: MCP服务器原则上可以将所有的操作转换称后台调度执行的任务与此相关能力包括list:服务器是否允许客户端查询当前正在运行或已完成的任务;cancel: 服务器是否支持远程终止任务requests服务器是否可以在任务执行过程中反向请求客户端虽然定义了这一系列能力但是FastMCP作为一个开箱即用的框架并不希望你去手工开启或者关闭某个能力所以并没有提供对服务器能力设置的API。客户端可以利用ClientSession的get_server_capabilities方法得到MCP服务器提供的能力上面提到的针对资源的订阅和解除订阅可以通过subscribe_resource和unsubscribe_resource方法来完成。订阅资源发送变化的通知可以通过设置Client的message_handler进行处理。classClientSession(BaseSession[types.ClientRequest,types.ClientNotification,types.ClientResult,types.ServerRequest,types.ServerNotification,]):defget_server_capabilities(self)-types.ServerCapabilities|Noneasyncdefsubscribe_resource(self,uri:AnyUrl)-types.EmptyResultasyncdefunsubscribe_resource(self,uri:AnyUrl)-types.EmptyResult2. 客户端能力客户端能力通过如下这个mcp.typesClientCapabilities类型来体现:classClientCapabilities(BaseModel):experimental:dict[str,dict[str,Any]]|NoneNonesampling:SamplingCapability|NoneNoneelicitation:ElicitationCapability|NoneNoneroots:RootsCapability|NoneNonetasks:ClientTasksCapability|NoneNonemodel_configConfigDict(extraallow)ClientCapabilities的字段体现了如下的能力experimental以字典的形式定自由定义实验性的特性samplingLLM采样功能。它允许我们工具执行期间向客户端LLM请求文本生成这使得工具能够利用客户端提供的AI功能进行分析、生成、推理等操作而无需客户端协调多次调用elicitation:处理服务器对结构化用户输入的请求。当工具在执行过程中需要向客户端请求用于输入时可以使用使用此功能。这使得MCP服务器无需预先获取所有输入而是可以交互式地询问缺失的参数、请求澄清或收集其他上下文信息;roots: 为MCP服务器当前执行的操作提供本地上下文和资源边界。当需要告知服务器客户端可以访问哪些本地资源时可以使用此功能。客户端利用提供的根目录roots用于告知服务器它可以提供的资源进而为操作框定一个边界tasks客户端支持对任务进行生命周期管理如显示进度显式、接收任务更新用于配合服务端的Tasks能力。由于所有的组件工具、资源和提示词最终都体现为一个具体的函数如果客户端请求的组件函数需要长时间执行可以以后台任务的形式调度执行这样客户端就可以避免长时间等待。此时服务器会立即响应请求并返回任务ID。客户端可以利用此ID追踪任务的状态并最终得到执行结果默认情况下客户端都不提供针对上述这些能力的支持。如果需要支持roots、sampling和elicitation字段体现的这些客户端能力需要调用如下这些方法。classClient(Generic[ClientTransportT],ClientResourcesMixin,ClientPromptsMixin,ClientToolsMixin,ClientTaskManagementMixin,):defset_roots(self,roots:RootsList|RootsHandler)-Nonedefset_sampling_callback(self,sampling_callback:SamplingHandler,sampling_capabilities:mcp.types.SamplingCapability|NoneNone,)-Nonedefset_elicitation_callback(self,elicitation_callback:ElicitationHandler)-None上述三个方法说明如下set_roots设置代表本地上下文和资源边界的“根目录”或者用于确定根目录的处理器RootsHandlerset_sampling_callback设置回调SamplingHandler用于处理MCP服务器发送的采样请求此回调也可以在Client的构造函数中利用sampling_handler参数指定set_elicitation_callback设置回调ElicitationHandler用于提供MCP服务器所需的结构化用户输入此回调也可以在Client的构造函数中利用elicitation_handler参数指定3. 初始化过程的能力交换服务端和客户之间针对各自能力交换是初始化Session的核心功能。我们接下来通过一个简单的演示实例看看在采用HTTP传输的情况下能力交换是如何利用Session初始化的请求/响应完成的。如下所示的是服务器程序并没有什么特别之处。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)客户端程序在根据MCP服务器地址将Client对象创建之后我们调用它的set_roots和set_sampling_callback方为它添加了动态执行资源边界和客户端采样的能力。importasynciofromfastmcpimportClient clientClient(transporthttp://localhost:3721/mcp)asyncdefmain():client.set_roots(roots[file://greet])client.set_sampling_callback(lambdamessages,params,context:...)asyncwithclient:passif__name____main__:asyncio.run(main())客户端程序启动后向对方发送的第一个请求就会携带自身的能力清单对方MCP服务器回复以自身的能力清单如下所示正是请求和响应的内容。请求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 Content-Length: 194 { method: initialize, params: { protocolVersion: 2025-11-25, capabilities: { sampling: {}, roots: { listChanged: true } }, clientInfo: { name: mcp, version: 0.1.0 } }, jsonrpc: 2.0, id: 0 }响应HTTP/1.1 200 OK date: Mon, 30 Mar 2026 01:03:13 GMT server: uvicorn cache-control: no-cache, no-transform connection: keep-alive content-type: text/event-stream mcp-session-id: 0e267706de3648979193203e32046791 x-accel-buffering: no Content-Length: 336 event: message data: { jsonrpc: 2.0, id: 0, result: { protocolVersion: 2025-11-25, capabilities: { experimental: {}, prompts: { listChanged: true }, resources: { subscribe: false, listChanged: true }, tools: { listChanged: true }, extensions: { io.modelcontextprotocol/ui: {} } }, serverInfo: { name: Greeting, version: 3.1.1 } } }顺便说一下旨在完成上述能力协商的初始化结束之后客户端还会发送一个初始化完成的确认请求JSON-RPC方法为notifications/initializedMCP服务器回复202 Accepted响应作为自己完成初始化的确认。具体的请求和响应内容如下所示请求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: 0e267706de3648979193203e32046791 mcp-protocol-version: 2025-11-25 Content-Length: 54 {method:notifications/initialized,jsonrpc:2.0}响应HTTP/1.1 202 Accepted date: Mon, 30 Mar 2026 01:03:13 GMT server: uvicorn content-type: application/json mcp-session-id: 0e267706de3648979193203e32046791 content-length: 0在默认情况下当我们以async with作用于Client对象的时候它会自动完成初始化流程。如果我们在调用Client的构造函数时将auto_initialize参数设置为False我们可以手工调用它的initialize方法自主完成初始化。该方法可以指定timeout参数控制调用超时事件返回的InitializeResult就包含服务端返回能力集。classClient(Generic[ClientTransportT],ClientResourcesMixin,ClientPromptsMixin,ClientToolsMixin,ClientTaskManagementMixin,):asyncdefinitialize(self,timeout:datetime.timedelta|float|int|NoneNone,)-mcp.types.InitializeResult:classInitializeResult(Result):protocolVersion:str|intcapabilities:ServerCapabilities serverInfo:Implementation instructions:str|NoneNone如下的程序演示了这种自主初始化的编程方式。顺便提一下虽然没有初始化但是当我们async with作用于Client完成上下文初始化后Client处理连接状态下面的演示程序也体现了这一点。importasynciofromfastmcpimportFastMCPfromfastmcpimportClientimportjson mcpFastMCP(Greeting)mcp.tool()asyncdefgreet(name:str)-str:Get a greeting message for the given namereturnfHi,{name}!clientClient(mcp,auto_initializeFalse)asyncdefmain():client.set_roots(roots[file://greet])client.set_sampling_callback(lambdamessages,params,context:...)asyncwithclient:assertclient.is_connected()assertclient.initialize_resultisNoneresultawaitclient.initialize(timeout5)print(json.dumps(result.model_dump(),indent2))if__name____main__:asyncio.run(main())输出{meta:null,protocolVersion:2025-11-25,capabilities:{experimental:{},logging:{},prompts:{listChanged:true},resources:{subscribe:false,listChanged:true},tools:{listChanged:true},completions:null,tasks:{list:{},cancel:{},requests:{tools:{call:{}},prompts:{get:{}},resources:{read:{}}}},extensions:{io.modelcontextprotocol/ui:{}}},serverInfo:{name:Greeting,title:null,version:0.0.0,websiteUrl:null,icons:null},instructions:null}

更多文章