买了两本MCP的书，每本三百多页，却没有一本把MCP说清楚的，感觉都是在凑字数。
好不容易读完，然后捋一下，总结这篇文章。
想学mcp的朋友，可以先看这篇文章，基本上能满足80％以上的需求。
一、什么是MCP
接触过agent编程的都知道，model在bind_tools后，tools是固定的，每次要新增一个tool，都要新建一个函数，然后修改传入bind_tools中的参数。也就是说，这是静态的、运行时不可变的。
但是，在实际的情况下，agent能力是可以动态增长的，特别是由agent动态编写脚本的能力，使得tool在运行时动态增加成为可能，或者另外一种情形，保持agent不重启的情况下增减能力，比如说开发者接到新的需求，开发好后部署这个能力，而agent在不重启的情况下知道有这个新的能力。
这种情况下，原有的bind_tools和静态编写tool function就做不到了。此时，mcp横空出世，为动态扩展agent能力提供了途径。
那到底mcp是什么？你只需要这样理解就行：
- 它就是agent tools
- 它提供了某种工作机制使得可以在运行时动态增、减tools
二、MCP架构与组件
MCP由下面三大组件组成：
- MCP Server：提供功能的服务端，也就是tool的实现端。
- MCP Client：用于访问MCP Server的客户端，一般以SDK形式存在。它负责从MCP Server中获取可用的tools和调用这些tools。由于它的存在，使用它的项目无需理会访问MCP Server的通信协议，只需要获得结果就行，使用起来就根原生的tools一样。
- Host：即使用MCP Client的应用，这里一般是ai agent。它将MCP Client SDK包含进项目，实例化一个MCP Client，然后通过它访问MCP Server。严格来说，Host并不属于MCP的一部分，它只是使用MCP的应用或者agent。
以上三个组件的架构如下图所示：
三、示例代码与讲解
了解到了MCP的组成和架构，那么我们用最简单的一个示例来讲解。
首先要将来面的pip加入项目中：
- mcp
- dotenv
- langgraph
- langchain-openai
- langchain-mcp-adapters
1. MCP Server实现
一般分为三步完成：
(1) 定义MCP Server的基本信息，如名字、监听端口
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
"Basic Usage MCP", port=9000
)
(2) 申明定义tool：
@mcp.tool(
name="call_mary",
description="Call Mary for work",
)
async def call_mary(message: str) -> str:
print(f'start calling Mary for work: {message}')
print(f'end calling Mary for work: {message}')
return 'success'
@mcp.tool(
name="pay",
description="Pay for work"
)
async def pay(who: str, money: float | int) -> str:
print(f'start paying Mary for work: {who}, {money}')
print(f'end paying Mary for work: {who}, {money}')
return 'success'
(3) 启动Server
MCP Server启动时，会自动读取所有@mcp.tool注解的tool，收集它们的信息，然后再启动一个Http Server。
def main() -> None:
try:
asyncio.run(mcp.run_sse_async())
except KeyboardInterrupt:
print('Begin stopping server by user operation')
if __name__ == '__main__':
main()
对，MCP Server实现起来就是这么简单。我们启动一下服务，看看日志输出：
$ python server.py
INFO: Started server process [76494]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:9000 (Press CTRL+C to quit)
完整代码如下：
server.py
import asyncio
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
"Basic Usage MCP", port=9000
)
@mcp.tool(
name="call_mary",
description="Call Mary for work",
)
async def call_mary(message: str) -> str:
print(f'start calling Mary for work: {message}')
print(f'end calling Mary for work: {message}')
return 'success'
@mcp.tool(
name="pay",
description="Pay for work"
)
async def pay(who: str, money: float | int) -> str:
print(f'start paying Mary for work: {who}, {money}')
print(f'end paying Mary for work: {who}, {money}')
return 'success'
def main() -> None:
try:
asyncio.run(mcp.run_sse_async())
except KeyboardInterrupt:
print('Begin stopping server by user operation')
if __name__ == '__main__':
main()
2. Host实现
为什么不是MCP Client实现？
从架构图上看，MCP Client只是以SDK的形式出现在Host中，所以我们只需要Host中实例化一个MCP Client的实例即可。
为了让整个项目能够跑起来，我们这里做一个简易的langgraph的agent。
(1) 导入相关的包
import asyncio
# 用于langgraph
from typing import Annotated
# 用于加载ENV
from dotenv import load_dotenv
# MCP Client SDK
from langchain_mcp_adapters.client import MultiServerMCPClient
# 用于langgraph
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END, START, add_messages
from langgraph.prebuilt import ToolNode
from pydantic import BaseModel
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
(2) 创建MCP Client，并获取对应的tools
步骤就是：
- 创建一个MCP Client
- 获取所有MCP Server所支持的tools
async def get_agent_tools():
"""
get agent tools
:return: tools and tool node
"""
mcp_client = MultiServerMCPClient({
"basic": {
"url": "http://localhost:9000/sse",
"transport": "sse"
}
})
return await mcp_client.get_tools()
(3) 组langgraph图，生成agent
- 创建model
这里和一般的创建model没什么区别，也是把tools传进去，然后bind_tools。
def get_model(tools):
"""
get model
:param tools: tools
:return: model
"""
return ChatOpenAI(
model="deepseek-v4-flash",
temperature=0,
extra_body={
# disable thinking
"thinking": {
"type": "disabled"
}
}
).bind_tools(tools)
- 组图并生成agent
这里也和一般的langgraph的用法没什么区别，就是与llm交互，进入tool_node，调用tool，把结果返回给llm而已。
不熟悉langgraph的读者可以先去学习langgraph，没办法，我目前暂时不熟悉其它的框架，只会langgraph。
def build_agent(model, tools):
workflow = StateGraph(AgentState)
workflow.add_node('tools', ToolNode(tools))
def call_model(state: AgentState):
messages = state.messages
response = model.invoke(messages)
return {
"messages": [response]
}
workflow.add_node('call_model', call_model)
# define should_continue edge
def should_continue(state:AgentState) -> str:
messages = state.messages
last_message = messages[-1]
if isinstance(last_message, AIMessage) and last_message.tool_calls:
return 'tools'
else:
return END
# connect each node
workflow.add_edge(START, "call_model")
workflow.add_conditional_edges("call_model", should_continue)
workflow.add_edge("tools", "call_model")
# build agent
return workflow.compile()
- 启动agent来完成任务
这里有一个让人疑惑的点，虽然创建了一个MCP Client，但是从上面的代码中并没有明显的调用client的代码，只是仅仅从调用它获取了tools。其实关键就在调用它获取tools这个调用中，它返回的tools并不是简单的tools,而是经过“包装增强”的tools，让llm调用它时会自动访问远端MCP Server，如果读者有兴趣，可以直接研读代码，但是对于实用主义者来说，知道它的工作原理即可。
async def main() -> None:
# 获取 tools
tools = await get_agent_tools()
# 获取 model
model = get_model(tools)
# 生成 agent
agent = build_agent(model, tools)
# 启动agent，注意，这里是`ainvoke`，不是`invoke`，因为tools用的是asyncio，所以这里需要明确调用`ainvoke`
state = await agent.ainvoke({
"messages": [
HumanMessage("Call Mary for work 'show me the money', and then pay her $3000")
]
})
print(state["messages"][-1].content)
if __name__ == '__main__':
load_dotenv(verbose=True)
asyncio.run(main())
完整代码如下：
host.py
import asyncio
from typing import Annotated
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END, START, add_messages
from langgraph.prebuilt import ToolNode
from pydantic import BaseModel
class AgentState(BaseModel):
"""
agent state
"""
messages: Annotated[list[BaseMessage], add_messages]
async def get_agent_tools():
"""
get agent tools
:return: tools and tool node
"""
mcp_client = MultiServerMCPClient({
"basic": {
"url": "http://localhost:9000/sse",
"transport": "sse"
}
})
return await mcp_client.get_tools()
def get_model(tools):
"""
get model
:param tools: tools
:return: model
"""
return ChatOpenAI(
model="deepseek-v4-flash",
temperature=0,
extra_body={
# disable thinking
"thinking": {
"type": "disabled"
}
}
).bind_tools(tools)
def build_agent(model, tools):
workflow = StateGraph(AgentState)
workflow.add_node('tools', ToolNode(tools))
def call_model(state: AgentState):
messages = state.messages
response = model.invoke(messages)
return {
"messages": [response]
}
workflow.add_node('call_model', call_model)
# define should_continue edge
def should_continue(state:AgentState) -> str:
messages = state.messages
last_message = messages[-1]
if isinstance(last_message, AIMessage) and last_message.tool_calls:
return 'tools'
else:
return END
# connect each node
workflow.add_edge(START, "call_model")
workflow.add_conditional_edges("call_model", should_continue)
workflow.add_edge("tools", "call_model")
# build agent
return workflow.compile()
async def main() -> None:
tools = await get_agent_tools()
model = get_model(tools)
agent = build_agent(model, tools)
state = await agent.ainvoke({
"messages": [
HumanMessage("Call Mary for work 'show me the money', and then pay her $3000")
]
})
print(state["messages"][-1].content)
if __name__ == '__main__':
load_dotenv(verbose=True)
asyncio.run(main())
我们启动它，然后查看日志来进行分析：
$ python host.py
Done! I've called Mary with the message "show me the money" and then paid her $3,000.
可以看到调用成功，并且正确说明了情况。
我们来分析一下MCP Server的日志：
# 这一块是MCP Client 向 MCP Server 获取可用的tools的日志
#
INFO: 127.0.0.1:58990 - "GET /sse HTTP/1.1" 200 OK
INFO: 127.0.0.1:58994 - "POST /messages/?session_id=da04a41c2ebd48f79dda2693dbea56e4 HTTP/1.1" 202 Accepted
INFO: 127.0.0.1:58994 - "POST /messages/?session_id=da04a41c2ebd48f79dda2693dbea56e4 HTTP/1.1" 202 Accepted
INFO: 127.0.0.1:58994 - "POST /messages/?session_id=da04a41c2ebd48f79dda2693dbea56e4 HTTP/1.1" 202 Accepted
Processing request of type ListToolsRequest
# 这一块是MCP Client 调用 MCP Server的tool，
# 与此同时再一次获取可用的tools
INFO: 127.0.0.1:59002 - "GET /sse HTTP/1.1" 200 OK
INFO: 127.0.0.1:59016 - "POST /messages/?session_id=ca534364f38e4ae6b8c493d6ba45c31b HTTP/1.1" 202 Accepted
INFO: 127.0.0.1:59016 - "POST /messages/?session_id=ca534364f38e4ae6b8c493d6ba45c31b HTTP/1.1" 202 Accepted
INFO: 127.0.0.1:59016 - "POST /messages/?session_id=ca534364f38e4ae6b8c493d6ba45c31b HTTP/1.1" 202 Accepted
start calling Mary for work: show me the money
end calling Mary for work: show me the money
INFO: 127.0.0.1:59016 - "POST /messages/?session_id=ca534364f38e4ae6b8c493d6ba45c31b HTTP/1.1" 202 Accepted
Processing request of type CallToolRequest
Processing request of type ListToolsRequest
# 这一块是MCP Client 调用 MCP Server的tool，
# 与此同时再一次获取可用的tools
INFO: 127.0.0.1:59024 - "GET /sse HTTP/1.1" 200 OK
INFO: 127.0.0.1:59030 - "POST /messages/?session_id=fb9ac6ec1ede4f5d8d74641d224b0828 HTTP/1.1" 202 Accepted
INFO: 127.0.0.1:59030 - "POST /messages/?session_id=fb9ac6ec1ede4f5d8d74641d224b0828 HTTP/1.1" 202 Accepted
INFO: 127.0.0.1:59030 - "POST /messages/?session_id=fb9ac6ec1ede4f5d8d74641d224b0828 HTTP/1.1" 202 Accepted
start paying Mary for work: Mary, 3000
end paying Mary for work: Mary, 3000
INFO: 127.0.0.1:59030 - "POST /messages/?session_id=fb9ac6ec1ede4f5d8d74641d224b0828 HTTP/1.1" 202 Accepted
Processing request of type CallToolRequest
Processing request of type ListToolsRequest
由Server端的日志可以得知，llm正确两次调用了MCP Server定义的tools，并且顺序都是对的。
另外得到一个很有用的信息，就是每次调用后都会自动重新获取一次tools，这也就是说明了mcp client会动态更新model里的tools，这样model就更新了他的tools列表，就达到了运行时更新tools的能力。
四、总结
MCP没有想像中那么复杂，但mcp的用法和思想却是相当考究的，什么时候用mcp，怎么用mcp，这里是一个大问题。
另外，还有一个slot的概念，即将消息上下文发给mcp server，让上下文可以延续。但是我是不支持这样用的，一个是数据安全问题，另一个则是这并不是mcp所应片处理的范围，它不应去理解调用它的agent上下文，它只需要按照需求得出结果即可,所以mcp在这上面的设计其实是过度的。
希望这篇文章能帮助到学习mcp的人。
写这个示例代码纠正了我一个长期以来的一个错误的观念，就是llm是直接调用tool的，其实不然，它是通过返回一个AIMessage让agent自行处理message的，所以，当model调用远端llm时，得到的message需要走一趟tool_node，也就是这个原因。
