前言
既然MCP
都已经出现了,甚至已经纳入面试题目了,就简单尝试一下这个新玩意儿。
建立一个最简单的stdio工具
我们可以从其他的一些博客中了解到,MCP
本质上就是一个MCP Server
配上一个MMCP Client>
,然后MCP Client
就可以调用MCP Server
提供的服务。
既然如此,我们的首要任务也就是建立一个相当简易的MCP Server
。
无参
既然要尽可能简单,那就输出Hello, World!
吧。
所以,我们的方法就可以定义出来:
def hello() -> str: return "Hello, World!"
为了让MCP
发现他是一个工具类方法,我们再加一个装饰器:
from mcp.server.fastmcp import FastMCPmcp = FastMCP("HelloServer")@mcp.tool(description="Say hello to the world")def hello() -> str: return "Hello, World!"if __name__ == "__main__": mcp.run()
看上去没啥问题,我们把这段写进server/basic_server.py
文件中。
通过查看源码,我们可以知道,我们没有指定host
(主机域名或IP)、port
(主机开放端口)、transport
(服务提供方式),于是MCP
会给我们分配一个默认的配置:
host
:127.0.0.1
port
:8000
transport
:stdio
也就是说,上述代码,我们创立了一个运行在命令行中的MCP
服务器。
然后,我们就可以在client
中调用MCP
了:
import asynciofrom mcp import ClientSession, StdioServerParameters, typesfrom mcp.client.stdio import stdio_clientasync def hello() -> None: # 通过 stdio 启动本地的 server.py server = StdioServerParameters( command="python", args=["servers/basic_server.py"], ) # 连接并初始化会话 async with stdio_client(server) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 调用 "hello" 工具 result = await session.call_tool("hello") # 打印文本类型的返回内容 for c in result.content: if isinstance(c, types.TextContent): print(c.text) breakif __name__ == "__main__": asyncio.run(hello())
可以看到,官方大量使用了async
进行异步传输,所以我们也将大量使用await
和asyncio
进行异步传输。
我们最后在执行这个客户端的时候,他会首先利用StdioServerParameters
从终端启动MCP Server
,然后再启动MCP Client
,最后进行异步交互。
在交互过程中,我们拿到所有的输出,然后选择我们需要的输出。按道理来说,这个案例里面只会输出TextContent
类型的结果,我们也只取其中的text
属性。
也就是说,在MCP Client
执行的终端里面,我们会看到Hello World!
。MCP Server
那边因为没有单独启动,所以什么都没有。
有参
既然无参弄完了,那就试试有参?
就比如说,我输入什么东西,他都会返回:Hello, <your-input>
。
说来也简单,就是每个都带上参数就好了:
from mcp.server.fastmcp import FastMCPmcp = FastMCP("HelloServer")@mcp.tool(description="Say hello to anything")def hello(text: str) -> str: return f"Hello, {text}!"if __name__ == "__main__": mcp.run()
整挺好。写进servers/param_server.py
中。
客户端也改一下:
import asynciofrom mcp import ClientSession, StdioServerParameters, typesfrom mcp.client.stdio import stdio_clientasync def hello(text: str) -> None: # 通过 stdio 启动本地的 server.py server = StdioServerParameters( command="python", args=["servers/param_server.py"], ) # 连接并初始化会话 async with stdio_client(server) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 调用 "hello" 工具 result = await session.call_tool("hello", {"text": text}) # 打印文本类型的返回内容 for c in result.content: if isinstance(c, types.TextContent): print(c.text) breakif __name__ == "__main__": asyncio.run(hello("LangChain"))
于是,客户端就会输出:Hello, LangChain!
。
改用HTTP传输方式
我们可以从其他的博客中,看到传输方式包含stdio
、sse
、http
三种。因为stdio
过于受限,sse
又逐步暴露出更多的缺点,所以接下来的案例就直接上http
了。
首先,因为默认给出来的就是stdio
,所以我们首先要改一下MCP Server
的传输方式配置,就像这样:
from mcp.server.fastmcp import FastMCPmcp = FastMCP("HelloServer")@mcp.tool(description="Say hello to anything")def hello(text: str) -> str: return f"Hello, {text}!"if __name__ == "__main__": mcp.run(transport="streamable-http")
虽然说,上面把transport
参数和host
、port
参数放在了一起,但实际上他们在不同的位置起作用。transport
参数在run
中指定,而host
与port
参数在FastMCP
构造函数中指定。
当然,我们还可以看源码,发现run(transport="streamable-http")
实际等同于run_streamable_http_async()
,因此我们还可以这么写:
from mcp.server.fastmcp import FastMCPmcp = FastMCP("HelloServer")@mcp.tool(description="Say hello to anything")def hello(text: str) -> str: return f"Hello, {text}!"if __name__ == "__main__": mcp.run_streamable_http_async()
值得注意的是,我们在其中并没有指定访问路由。这是因为MCP Server
默认访问路由就是http://127.0.0.1:8000/mcp/
。如果需要改动,同样在构造函数中指明:
mcp = FastMCP( name = "HelloServer", # 名称,可以为None host = "localhost", # 默认值 port = 8000, # 默认值 streamable_http_path = "/mcp" # 默认值)
然后客户端的改动说大也不大:
import asynciofrom mcp import ClientSession, typesfrom mcp.client.streamable_http import streamablehttp_clientasync def hello(text: str): # Connect to HTTP streaming server async with streamablehttp_client("http://localhost:8000/mcp/") as (read, write, get_session_id_callback): async with ClientSession(read, write) as session: await session.initialize() # Call "hello" tool result = await session.call_tool("hello", {"text": text}) # Print text-type return content for c in result.content: if isinstance(c, types.TextContent): print(c.text) breakif __name__ == "__main__": asyncio.run(hello("LangChain"))
这里值得注意的是,streamablehttp_client
包含三个内容:MemoryObjectReceiveStream
、MemoryObjectSendStream
和GetSessionIdCallback
,从字面意义上区分也就是read
、write
和get_session_id_callback
。而ClientSession
的构造函数中,大量参数与上述三个的交集只有两个,分别是MemoryObjectReceiveStream
和MemoryObjectSendStream
,也就是只需要read
和write
。
剩下的就没变化了。
日志解析
虽然说在客户端这边只有一个Hello Langchain!
,但是由于HTTP
请求会单独拉起一个MCP Server
,所以在MCP Server
日志中,会产生一些日志:
INFO: 127.0.0.1:37280 - "POST /mcp/ HTTP/1.1" 307 Temporary Redirect[08/13/25 15:55:33] INFO Created new transport with session ID: 6cf4fa2b3f8941cba2aff10439e268f9 streamable_http_manager.py:233INFO: 127.0.0.1:37280 - "POST /mcp HTTP/1.1" 200 OKINFO: 127.0.0.1:37294 - "POST /mcp/ HTTP/1.1" 307 Temporary RedirectINFO: 127.0.0.1:37308 - "GET /mcp/ HTTP/1.1" 307 Temporary RedirectINFO: 127.0.0.1:37294 - "POST /mcp HTTP/1.1" 202 AcceptedINFO: 127.0.0.1:37308 - "GET /mcp HTTP/1.1" 200 OKINFO: 127.0.0.1:37322 - "POST /mcp/ HTTP/1.1" 307 Temporary RedirectINFO: 127.0.0.1:37322 - "POST /mcp HTTP/1.1" 200 OK INFO Processing request of type CallToolRequest server.py:625INFO: 127.0.0.1:37328 - "POST /mcp/ HTTP/1.1" 307 Temporary RedirectINFO: 127.0.0.1:37328 - "POST /mcp HTTP/1.1" 200 OK INFO Processing request of type ListToolsRequest server.py:625INFO: 127.0.0.1:37344 - "DELETE /mcp/ HTTP/1.1" 307 Temporary Redirect INFO Terminating session: 6cf4fa2b3f8941cba2aff10439e268f9 streamable_http.py:630INFO: 127.0.0.1:37344 - "DELETE /mcp HTTP/1.1" 200 OK
这么大一串,其实看下来就是这么个流程:
- 首先,默认服务是
/mcp
,你请求到了/mcp/
,没事,给你掰回去(Redirecting
);掰回去了,对上号了,用GET
请求给你发一个号牌(SessionId
);然后,收到了一个POST
请求:CallToolRequest
;根据请求,先查一下字典里有没有这项服务:ListToolsRequest
;执行请求并返回;结束会话(DELETE
)这一套流程走完之后,一个依赖HTTP
的MCP
请求就这样结束了。