资料
MCP 官方文档
https://modelcontextprotocol.io/introduction
各个 clients 对 MCP 的支持情况
https://modelcontextprotocol.io/clients
MCP Python SDK：MCP Client 和 Server 官方 SDK
https://github.com/modelcontextprotocol/python-sdk
前言
MCP 客户端（MCP Clients）：在主机程序内部，与 MCP server 保持 1:1 的连接。比如支持配置 MCP 的客户端，底层都实现了 MCP Client，然后提供界面让用户配置 MCP Server，比如 Cursor、Claude、Cline、Codebuddy、AI 工具等等。
上面的流程图介绍了 MCP 跟 LLM 的完整交互过程，App 为支持调用 MCP Server 的客户端，比如 Cursor、Claude、Codebuddy 等，他们底层其实就是实现一个 MCP Client，有连接 MCP Server 和调用 Tool的能力。
接下来我会开发一个支持连接 MCP Server 的客户端，以及如何将 MCP Server Tools 提供给大模型，让模型调用 Tools，最终实现通过自然语言调用 MCP Server。
MCP Client Demo 开发
一、环境配置
- 开发语言：Python 3.13.2
- 操作系统：MacOS 或 Linux
- Python 项目管理工具：uv
- 框架：langchain
项目初始化
git clone https://github.com/zhengjianhong001/ai_agent_test
cd ai_agent_test
# 创建虚拟环境
uv venv && source .venv/bin/activate
# 安装相关依赖
uv pip install -r requirements.txt创建 .evn 配置环境变量
# OpenAI 配置，也可以填写deepseek或元宝的
OPENAI_API_KEY=your_openai_api_key_here
OPENAI_MODEL=gpt-4o-mini
OPENAI_API_BASE=https://api.openai.com/v1二、编写 Client 代码
通过本地的MCP Server Stdio调用
# Create server parameters for stdio connection
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
from langchain_core.messages import HumanMessage, ToolMessage
import asyncio
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents import AgentExecutor, create_openai_tools_agent
from dotenv import load_dotenv
import os
# 加载环境变量
load_dotenv()
# 把工具列表发送给 LLM，根据 LLM 返回需要调用的工具，然后手动调用工具，并把工具结果发送给模型，模型返回最终结果
base_params = {
"model": os.getenv("OPENAI_MODEL", ""),
"temperature": 0.1,
"streaming": True,
"openai_api_key": os.getenv("OPENAI_API_KEY", ""),
"openai_api_base": os.getenv("OPENAI_API_BASE", ""),
}
model = ChatOpenAI(**base_params)
server_params = StdioServerParameters(
command="uvx",
# Make sure to update to the full absolute path to your math_server.py file
args=["mcp-server-tapd"],
env={
"TAPD_ACCESS_TOKEN": os.getenv("TAPD_ACCESS_TOKEN", ""),
"TAPD_API_BASE_URL": os.getenv("TAPD_API_BASE_URL", ""),
"TAPD_BASE_URL": os.getenv("TAPD_BASE_URL", ""),
"CURRENT_USER_NICK": os.getenv("CURRENT_USER_NICK", ""),
"BOT_URL": os.getenv("BOT_URL", "")
}
)
async def run_agent():
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# Initialize the connection
await session.initialize()
# Get tools
tools = await load_mcp_tools(session)
# 直接绑定工具到模型
model_with_tools = model.bind_tools(tools)
# 第一轮：调用模型
messages = [HumanMessage(content="获取我参与的项目，只返回项目名称列表，不要返回其他内容")]
result = await model_with_tools.ainvoke(messages)
# 检查是否有工具调用
if result.tool_calls:
print("检测到工具调用，正在执行...")
# 执行工具调用
tool_results = []
for tool_call in result.tool_calls:
tool_name = tool_call['name']
tool_args = tool_call['args']
# 找到对应的工具
for tool in tools:
if tool.name == tool_name:
print(f"执行工具: {tool_name}")
tool_result = await tool.ainvoke(tool_args)
tool_results.append(tool_result)
break
# 第二轮：将工具结果发送给模型
if tool_results:
# 添加工具结果到消息列表
new_messages = messages + [result] + [
ToolMessage(
content=str(tool_results[0]), # 假设只有一个工具调用
tool_call_id=result.tool_calls[0]['id']
)
]
print("将工具结果发送给模型...")
final_result = await model_with_tools.ainvoke(new_messages)
return final_result
return result
# Run the async function
if __name__ == "__main__":
result = asyncio.run(run_agent())
print("AI 最终内容：", result.content)
# print(result)三、运行服务
python client.py
如果把两轮调用LLM 的参数打印出来，可以看到详细的过程
第一轮让 LLM 觉得使用哪个 Tool
content=''
additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_zmize4r6', 'function': {'arguments': '{"nick":""}', 'name': 'get_user_participant_projects'}, 'type': 'function'}]} response_metadata={'finish_reason': 'tool_calls', 'model_name': 'deepseek-v3'} id='run--c45512d8-c428-470d-acff-f951afe8e8' tool_calls=[{'name': 'get_user_participant_projects', 'args': {'nick': ''}, 'id': 'call_zmize4r6', 'type': 'tool_call'}] usage_metadata={'input_tokens': 79590, 'output_tokens': 135, 'total_tokens': 79725, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}- content=”
表示模型没有直接生成自然语言回答，而是选择调用工具，所以内容为空。 - additional_kwargs
包含工具调用的详细信息：tool_calls: 是一个列表，包含本次调用的工具信息index: 0: 工具调用的索引（如果多次调用则递增）id: 'call_zmize4r6': 本次工具调用的唯一标识function: 描述要调用的函数arguments: '{"nick":""}': 传递给函数的参数（这里nick为空字符串）name: 'get_user_participant_projects': 要调用的函数名称（推测是 “获取用户参与的项目”）
type: 'function': 调用类型为函数
- response_metadata
模型响应的元数据：finish_reason: 'tool_calls': 模型结束生成的原因是需要调用工具model_name: 'deepseek-v3': 使用的模型名称
- id: ‘run–c45512d8-c428-470d-acff-f951afe8e8’
本次模型运行的唯一标识 - tool_calls
与additional_kwargs中的tool_calls信息一致，是对工具调用的简化描述 - usage_metadata
令牌（token）使用情况：input_tokens: 79590: 输入的令牌数output_tokens: 135: 输出的令牌数total_tokens: 79725: 总令牌数- 其他字段描述了令牌的详细使用分类
所以，这个结果表示模型决定调用get_user_participant_projects函数来获取信息
第二轮将 Tool 结果再次返回给 LLM
content='以下是你参与的项目名称列表：\n\n1. test6\n2. test'
additional_kwargs={}
response_metadata={'finish_reason': 'stop', 'model_name': 'deepseek-v3'}
id='run--1deecf07-c77b-451b-8e1a-46849d476'
usage_metadata={'input_tokens': 15742830, 'output_tokens': 90950, 'total_tokens': 15833780, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}源码：
https://github.com/zhengjianhong001/ai_agent_test/tree/main/mcp