掘金 人工智能 前天 19:11
MCP入门级简单尝试
index_new5.html
../../../zaker_core/zaker_tpl_static/wap/tpl_guoji1.html

 

本文详细介绍了如何使用MCP框架构建和服务调用工具。首先,通过构建一个简单的无参和有参的stdio工具,演示了MCP Server的基本搭建和Client的调用流程,强调了异步编程在其中的应用。随后,文章重点转向HTTP传输方式,展示了如何配置MCP Server以支持streamable-http,以及客户端如何连接并调用HTTP服务。通过分析HTTP请求的日志,揭示了从请求发起、会话建立、工具调用到会话结束的完整生命周期,为读者提供了MCP框架在不同传输模式下的实践指南。

### 🚀 MCP框架基础:stdio传输模式下的工具构建与调用 MCP框架的核心在于其Server-Client架构,通过FastMCP类可以轻松创建MCP Server。文章首先以“Hello, World!”为例,展示了如何定义一个无参的工具方法,并使用`@mcp.tool`装饰器将其注册到Server中。接着,通过一个带参数的工具示例,说明了如何处理输入参数并进行返回。这部分内容强调了使用`asyncio`和`await`进行异步通信,以及`StdioServerParameters`在启动本地Server和Client交互中的作用,为理解MCP的通信机制奠定了基础。

### 🌐 转向HTTP传输:配置MCP Server与客户端的连接 文章指出stdio传输方式的局限性,并详细介绍了如何将MCP Server的传输方式切换为HTTP。通过在`mcp.run()`方法中指定`transport='streamable-http'`或直接调用`run_streamable_http_async()`,可以启用HTTP服务。同时,文章解释了如何通过`FastMCP`构造函数配置主机、端口和访问路由(如`/mcp/`)。客户端也相应地从`stdio_client`切换到`streamablehttp_client`,并演示了如何建立HTTP连接、初始化会话以及调用工具,展示了HTTP传输方式的灵活性。

### 🔍 日志解析:深入理解HTTP请求的生命周期 为了更清晰地说明HTTP传输的工作流程,文章对MCP Server的日志进行了详细解读。日志显示了一个典型的HTTP请求生命周期,包括:客户端向默认路由`/mcp/`发起请求,Server进行重定向到`/mcp/`;Server通过GET请求为客户端分配Session ID;客户端发送包含`CallToolRequest`的POST请求;Server通过`ListToolsRequest`查找可用工具;执行工具并返回结果;最后,通过`DELETE`请求结束会话。这一过程揭示了MCP框架在HTTP通信中的请求处理机制和状态管理。

### 💡 异步编程是关键:MCP框架的通信基础 无论是stdio还是HTTP传输,MCP框架都大量依赖Python的`asyncio`库进行异步操作。从Client端初始化会话、调用工具,到Server端的请求处理,都贯穿着`async`和`await`的使用。这使得MCP能够高效地处理并发请求,避免阻塞,从而提升整体性能。文章通过展示客户端和服务端的异步代码实现,强调了理解和运用异步编程是成功使用MCP框架的关键。

前言

既然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会给我们分配一个默认的配置:

也就是说,上述代码,我们创立了一个运行在命令行中的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进行异步传输,所以我们也将大量使用awaitasyncio进行异步传输。

我们最后在执行这个客户端的时候,他会首先利用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传输方式

我们可以从其他的博客中,看到传输方式包含stdiossehttp三种。因为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参数和hostport参数放在了一起,但实际上他们在不同的位置起作用。transport参数在run中指定,而hostport参数在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包含三个内容:MemoryObjectReceiveStreamMemoryObjectSendStreamGetSessionIdCallback,从字面意义上区分也就是readwriteget_session_id_callback。而ClientSession的构造函数中,大量参数与上述三个的交集只有两个,分别是MemoryObjectReceiveStreamMemoryObjectSendStream,也就是只需要readwrite

剩下的就没变化了。

日志解析

虽然说在客户端这边只有一个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

这么大一串,其实看下来就是这么个流程:

这一套流程走完之后,一个依赖HTTPMCP请求就这样结束了。

Fish AI Reader

Fish AI Reader

AI辅助创作,多种专业模板,深度分析,高质量内容生成。从观点提取到深度思考,FishAI为您提供全方位的创作支持。新版本引入自定义参数,让您的创作更加个性化和精准。

FishAI

FishAI

鱼阅,AI 时代的下一个智能信息助手,助你摆脱信息焦虑

联系邮箱 441953276@qq.com

相关标签

MCP框架 Python 异步编程 HTTP传输 工具调用
相关文章