MCP-Chinese-Getting-Started-Guide MCP工具

安装依赖

uv add "mcp[cli]" httpx openai

然后我们来创建一个叫 `web_search.py` 文件，来实现我们的服务。MCP 为我们提供了2个对象：`mcp.server.FastMCP` 和 `mcp.server.Server`，`mcp.server.FastMCP` 是更高层的封装，我们这里就来使用它。

Model Context Protocol(MCP) 编程极速入门

[TOC]

借助 serverless 将 MCP 服务部署到云端

上面我们讲的都是如何使用本地的 MCP 服务，但是有时我们希望直接把 MCP 服务部署到云端来直接调用，就省去了本地下载启动的烦恼了。此时，我们就需要来使用 MCP 的 SSE 的协议来实现了。

此时，我们先来写 SSE 协议的 MCP 服务。实现起来很简单，只需要将我们最后的 run 命令中的 transport 参数设置为 sse 即可。下面还是以上面的网络搜索为例子，来实现一下 ，具体代码如下：

```python

在 LangChain 中使用 MCP 服务器

最近 LangChain 发布了一个新的开源项目 langchain-mcp-adapters，可以很方便的将 MCP 服务器集成到 LangChain 中。下面我们来看看如何使用它:

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent

from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o")

server_params = StdioServerParameters(
    command='uv',
    args=['run', 'web_search.py'],
)

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()

        # 获取工具列表
        tools = await load_mcp_tools(session)

        # 创建并使用 ReAct agent
        agent = create_react_agent(model, tools)
        agent_response = await agent.ainvoke({'messages': '杭州今天天气怎么样？'})

更详细的使用方法请参考：https://github.com/langchain-ai/langchain-mcp-adapters

为 stdio 连接创建服务器参数

server_params = StdioServerParameters( # 服务器执行的命令，这里我们使用 uv 来运行 web_search.py command='uv', # 运行的参数 args=['run', 'web_search.py'], # 环境变量，默认为 None，表示使用当前环境变量 # env=None )

async def main(): # 创建 stdio 客户端 async with stdio_client(server_params) as (stdio, write): # 创建 ClientSession 对象 async with ClientSession(stdio, write) as session: # 初始化 ClientSession await session.initialize()

# 列出可用的工具 response = await session.list_tools() print(response)

# 调用工具 response = await session.call_tool('web_search', {'query': '今天杭州天气'}) print(response)

if name == 'main': asyncio.run(main())

因为我们的python脚本需要在虚拟环境中才能运行，所以这里我们通过 `uv` 来启动我们的脚本。

下面我们来通过一个小例子来看看如何让 `DeepSeek` 来调用我们 MCP 服务器中的方法。

这里我们会用 `dotenv` 来管理我们相关的环境变量。.env 文件内容如下：

首先我们来编写我们的 `MCPClient` 类。

from openai import OpenAI from dotenv import load_dotenv

from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client

load_dotenv()

class MCPClient: def init(self): self.session: Optional[ClientSession] = None self.exit_stack = AsyncExitStack() self.client = OpenAI()

然后我们添加 `connect_to_server` 方法来初始化我们的 MCP 服务器的 session。

stdio_transport = await self.exit_stack.enter_async_context( stdio_client(server_params)) stdio, write = stdio_transport self.session = await self.exit_stack.enter_async_context( ClientSession(stdio, write))

await self.session.initialize()

然后我们再实现一个用于调用 MCP 服务器的方法来处理和 DeepSeek 之间的交互。

# 获取所有 mcp 服务器 工具列表信息 response = await self.session.list_tools() # 生成 function call 的描述信息 available_tools = [{ "type": "function", "function": { "name": tool.name, "description": tool.description, "input_schema": tool.inputSchema } } for tool in response.tools]

# 请求 deepseek，function call 的描述信息通过 tools 参数传入 response = self.client.chat.completions.create( model=os.getenv("OPENAI_MODEL"), messages=messages, tools=available_tools )

# 处理返回的内容 content = response.choices[0] if content.finish_reason == "tool_calls": # 如何是需要使用工具，就解析工具 tool_call = content.message.tool_calls[0] tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments)

# 执行工具 result = await self.session.call_tool(tool_name, tool_args) print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n") # 将 deepseek 返回的调用哪个工具数据和工具执行完成后的数据都存入messages中 messages.append(content.message.model_dump()) messages.append({ "role": "tool", "content": result.content[0].text, "tool_call_id": tool_call.id, })

# 将上面的结果再返回给 deepseek 用于生产最终的结果 response = self.client.chat.completions.create( model=os.getenv("OPENAI_MODEL"), messages=messages, ) return response.choices[0].message.content

return content.message.content

接着，我们来实现循环提问和最后退出后关闭session的操作。

if query.lower() == 'quit': break

response = await self.process_query(query) print("\n" + response)

except Exception as e: import traceback traceback.print_exc()

async def cleanup(self): """Clean up resources""" await self.exit_stack.aclose()

最后，我们来完成运行这个客户端相关的代码

if name == "main": import sys

asyncio.run(main())

```

这是一个最精简的代码，里面没有实现记录上下文消息等功能，只是为了用最简单的代码来了解如何通过大语言模型来调动 MCP 服务器。这里只演示了如何连接单服务器，如果你期望连接多个 MCP 服务器，无非就是循环一下 connect_to_server 中的代码，可以将他们封装成一个类，然后将所有的 MCP 服务器中的工具循环遍历生成一个大的 available_tools，然后在通过大语言模型的返回结果进行调用即可，这里就不再赘述了。 > 可以参考官方案例：https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-chatbot/mcp_simple_chatbot/main.py

第一个参数会被传入上下文对象

async def web_search(ctx: Context, query: str) -> str: """ 搜索互联网内容

Args: query: 要搜索内容

Returns: 搜索结果的总结 """ # 如果之前问过同样的问题，就直接返回缓存 histories = ctx.request_context.lifespan_context.histories if query in histories： return histories[query]

async with httpx.AsyncClient() as client: response = await client.post( 'https://open.bigmodel.cn/api/paas/v4/tools', headers={'Authorization': 'YOUR API KEY'}, json={ 'tool': 'web-search-pro', 'messages': [ {'role': 'user', 'content': query} ], 'stream': False } )

res_data = [] for choice in response.json()['choices']: for message in choice['message']['tool_calls']: search_results = message.get('search_result') if not search_results: continue for result in search_results: res_data.append(result['content'])

return_data = '\n\n\n'.join(res_data)

# 将查询值和返回值存入到 histories 中 ctx.request_context.lifespan_context.histories[query] = return_data return return_data

if name == "main": app.run()

```