1.前言

今年以来 mcp实在太火了，有个比喻挺贴切的，当大模型有了 mcp就相当于有了手和脚，真正可以替用户干活了。甚至，有预言 mcp会是未来专属大模型的 app。

故接函数调用后，接着用 langchain实现 mcp调用，环境信息如下：

​ python-3.13、langchain-0.3.26

2.创建MCP Server

通过对比，最终选取比较轻量化的框架 fastmcp来构建 mcp server。

requirements.txt内容如下：

fastmcp==2.8.0

首先，创建 mcp实例，并定义工具函数：

mcp = FastMCP("MCP-Server-Demo", port=9698)

@mcp.tool()
def get_weather(location: str):
	"""获取指定城市的天气信息"""
    return f"{location} 的天气是晴天，22°C"

@mcp.tool()
def get_time(location: str):
	"""获取指定城市的时间"""
    return f"{location} 的当前时间是 9:00 PM"

@mcp.tool()注解会把普通 python函数包装成一个 mcp工具。

mcp server常见的通讯方式有三种：stdio, streamable-http, sse。这里选择以 sse的方式启动：

if __name__ == "__main__":
    # 使用SSE传输方式启动服务器
    mcp.run(transport="sse", host="0.0.0.0")

3.创建MCP 客户端

另外新建一个使用 langchain的项目。

requirements.txt内容如下：

requests==2.32.4
langchain==0.3.26
openai==1.93.0
langchain-community==0.3.26
langchain-openai==0.3.27
langgraph==0.5.0
langchain-mcp-adapters==0.1.7

在 langchain里面创建 mcp客户端比较简单，把第 2步的 mcp地址填一下：

mcp_client = MultiServerMCPClient(
    {
        "task": {
            "url": "http://127.0.0.1:9698/sse",
            "transport": "sse",
        }
    }
)

最后，定义获取 mcp服务端工具列表的方法：

async def get_mcp_tools():
    return await mcp_client.get_tools()

mcp_tools = asyncio.run(get_mcp_tools())

get_tools是一个异步函数，可以获取在第2步中使用 @mcp.tool()定义的函数，并获取函数的注释作为工具的的描述信息。比如 get_weather: 获取指定城市的天气信息，模型就会根据这个描述信息来选择适合用户问题的工具。

4.构建Agent

首先，初始化大模型：

llm_cfg = {
    "base_url": "https://api.siliconflow.cn/v1",
    "api_key": "sk-xx",
}

model = init_chat_model(
    model="Qwen/Qwen2.5-72B-Instruct", model_provider="openai", **llm_cfg
)

这里的模型只要兼容 openai协议即可。

其次，编写 system prompt，这个很有必要，可以给模型划定行业、确定领域、执行步骤和回答格式等，如下：

SYS_PROMPT = """你是一个气象领域的专家，擅长分析城市天气，并给予出行建议。
分析思路：
  1.先分析用户的问题，获取城市信息
  2.根据用户问题选择要调用的工具
回答格式如下：
  1.输出要调用的工具
  2.根据工具调用情况总结，并回复用户
"""

然后，构建 agent执行器：

memory = MemorySaver()

agent_executor = create_react_agent(
    model, mcp_tools, prompt=SYS_PROMPT, checkpointer=memory
)

最后，执行 agent：

config = {"configurable": {"thread_id": "abc123"}}

input_message = {
    "role": "user",
    "content": "成都的天气怎么样，现在几点了",
}


async def main():
    async for step, metadata in agent_executor.astream(
        {"messages": [input_message]}, config, stream_mode="messages"
    ):
        if isinstance(step, ToolMessage):
            print("\n--------------------------------------------")
            print(f"Tool: {step.name}, result: {step.content}")
            print("--------------------------------------------")
        else:
            print(step.text(), end="", flush=True)


if __name__ == "__main__":
    asyncio.run(main())

实现效果如下：

首先，需要查询天气查询工具获取成都的天气信息

Tool: get_weather, result: 成都的天气是晴天，22°C

其次，通过时间工具获取成都的时间

Tool: get_time, result: 成都的当前时间是 9:00 PM

成都现在是9:00 PM，天气是晴天，温度为22°C，适合外出，进行各项室外活动。

5.总结

mcp确实会是未来增强大模型能力的一种极佳选择，后续不同团队开发的服务预计会有两种形态，一种是给传统应用提供的 restful服务，另外一种则是给 agent进行调用的 mcp服务。另外，项目落地过程中，system prompt会对工具调用的准确性和稳定性具有决定性作用，要特别关注。还有，就是 agent这种方式会进行多轮调用，每轮还会带上之前的内容，对 token的消耗速度堪称恐怖。