摘要
随着 AI Agent 能力的不断增强，如何让大模型具备可控、可扩展的工具调用能力，成为工程落地中的关键问题。MCP（Model Context Protocol）作为一种新兴的 Agent 与工具通信协议，为构建标准化、可复用的智能系统提供了新的思路。
本文将从零开始，使用 Python 搭建一个完整的 MCP Server，并结合 LangChain 实现 Agent 与工具的自动注册、调度与调用。内容涵盖 MCP Server 的基础结构设计、工具暴露方式、Agent 调用流程以及关键实现细节，通过完整的实战代码，帮助读者理解 “Agent 如何通过 MCP 使用工具” 的完整闭环。
文章适合具备一定 Python 基础、对 AI Agent、LangChain 或 MCP 协议感兴趣的开发者阅读，可作为 MCP Server 工程实践与 Agent 系统设计的入门参考。
运行效果
概述
这个项目主要包含以下几个核心组件：
核心代码解析
1. MCP服务器实现（pymcp.py）
# pymcp.py
# 主 MCP 服务器入口：从标准输入读取请求，调用工具并返回结果
from mcp_api import TOOLS, send_response
import sys
import json
from util.logger import log # 日志记录工具
def run_server():
"""运行 MCP 服务器主循环"""
for line in sys.stdin:
# 跳过空行
if not line.strip():
continue
log(f"收到请求: {line.strip()}")
# 解析 JSON 请求
try:
req = json.loads(line)
except Exception as e:
log(f"JSON 格式错误: {e}")
send_response(None, error=f"无效的 JSON: {e}")
continue
req_id = req.get("id")
method = req.get("method")
params = req.get("params", {})
# 初始化连接（客户端首次握手）
if method == "initialize":
# 构建工具能力对象（LM Studio 要求格式为对象）
tools_obj = {}
for key, meta in TOOLS.items():
tools_obj[meta["name"]] = {
"description": meta["description"],
"inputSchema": meta["inputSchema"]
}
send_response(req_id, result={
"protocolVersion": "2025-06-18",
"serverInfo": {
"name": "mcp-server-python",
"version": "0.1.0"
},
"capabilities": {
"tools": tools_obj
}
})
continue
# 调用工具（MCP 协议方法：tools/call）
if method == "tools/call":
tool_name = params.get("name")
arguments = params.get("arguments", {})
if tool_name not in TOOLS:
log(f"未知工具: {tool_name}")
send_response(req_id, error=f"未知工具: {tool_name}")
continue
try:
log(f"正在调用工具 {tool_name}，参数: {arguments}")
result = TOOLS[tool_name]["func"](**arguments)
log(f"工具 {tool_name} 返回: {result}")
# 按照 MCP 协议返回 content 数组
send_response(req_id, result={
"content": [
{
"type": "text",
"text": str(result)
}
]
})
except Exception as e:
log(f"工具 {tool_name} 执行出错: {e}")
send_response(req_id, error=str(e))
continue
# 客户端初始化完成通知（可忽略）
if method == "notifications/initialized":
# 这是客户端发来的通知，服务器无需回复！
log("收到 initialized 通知，忽略")
continue # 不要 send_response！
# 列出所有工具（兼容旧协议）
if method == "tools/list":
tools_info = []
for key, meta in TOOLS.items():
tools_info.append({
"name": meta["name"],
"description": meta["description"],
"inputSchema": meta["inputSchema"]
})
send_response(req_id, result={"tools": tools_info})
continue
# 忽略其他通知类消息（如 notifications/progress 等）
if method and method.startswith("notifications/"):
# 通知类消息不需要响应，但为安全起见返回一个空确认
send_response(None, result={"status": "ok"})
continue
# 其他未支持的方法
log(f"收到未知方法: {method}")
send_response(req_id, error=f"未知方法: {method}")
if __name__ == "__main__":
run_server()
这段代码是MCP服务器的核心实现，它监听标准输入中的JSON-RPC请求，并根据请求类型执行相应的操作。当收到initialize请求时，服务器会返回可用的工具列表；当收到tools/call请求时，服务器会调用指定的工具并返回结果。
2. MCP API接口定义（mcp_api.py）
# mcp_api.py
from tools.get_date import get_date
from tools.time_tool import get_current_time
import json
import sys
# 只注册一个工具：获取当前日期
TOOLS = {
"get_current_time": {
"name": "get_current_time",
"description": "获取当前时间，支持自定义格式或 ISO 8601 标准格式",
"inputSchema": {
"type": "object",
"properties": {
"format": {
"type": "string",
"description": "时间格式，例如 '%H:%M'、'%H时%M分'，或 'iso' 表示 ISO 8601",
"default": "%H:%M:%S"
}
},
"required": []
},
"func": get_current_time
},
"get_date": {
"name": "get_date",
"description": "获取当前日期，支持自定义格式（如 '%Y年%m月%d日'）",
"inputSchema": {
"type": "object",
"properties": {
"format": {
"type": "string",
"description": "日期格式，例如 '%Y-%m-%d'、'%Y年%m月%d日' 等",
"default": "%Y-%m-%d"
}
},
"required": []
},
"func": get_date
}
}
def send_response(req_id, result=None, error=None):
"""
发送符合 JSON-RPC 2.0 标准的响应
"""
response = {
"jsonrpc": "2.0", # ⬅️ 关键！必须加上
"id": req_id
}
if error is not None:
response["error"] = {
"code": –32000, # 自定义错误码
"message": str(error)
}
else:
response["result"] = result
# 输出到 stdout（MCP 协议要求）
print(json.dumps(response, ensure_ascii=False, separators=(',', ':')))
sys.stdout.flush()
这个文件定义了可用的工具以及响应发送函数。TOOL字典包含了两个工具：get_current_time和get_date，每个工具都有名称、描述、输入模式和对应的函数。
3. 工具实现
获取日期工具（tools/get_date.py）
# tools/get_date.py
from datetime import datetime
def get_date(format: str = "%Y-%m-%d") –> str:
"""
Returns the current date as a string.
Args:
format (str): Date format string (default: "%Y-%m-%d")
Returns:
str: Formatted current date
"""
return datetime.now().strftime(format)
获取时间工具（tools/time_tool.py）
# tools/time_tool.py
from datetime import datetime
def get_current_time(format: str = "%H:%M:%S") –> str:
"""
获取当前时间，支持自定义格式。
Args:
format (str): 时间格式字符串，默认为 "%H:%M:%S"
示例：
– "%H:%M" → "14:30"
– "%H时%M分%S秒" → "14时30分45秒"
– "iso" → ISO 8601 格式（如 "2026-01-16T14:30:45.123456"）
Returns:
str: 格式化后的时间字符串
"""
if format == "iso":
return datetime.now().isoformat()
else:
return datetime.now().strftime(format)
4. 日志工具（util/logger.py）
# util/logger.py
# 简单日志工具：将日志打印到 stderr，避免干扰 MCP 的 stdout 通信
import sys
from datetime import datetime
def log(message: str):
"""
记录日志到标准错误（stderr），带时间戳。
参数:
message (str): 要记录的日志内容
"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] {message}", file=sys.stderr, flush=True)
5. 主聊天代理（demo.py）
#!/usr/bin/env python3
"""
最终稳定版 MCP Chat Agent（支持 mcp.json）
– 自动加载 mcp.json 配置
– 启动你自己的 pymcp.py 作为 MCP 服务器
– 支持 Ollama / OpenAI 二选一
"""
import asyncio
import json
import os
from typing import Annotated, TypedDict
from langchain_ollama import ChatOllama
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, START
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph.message import add_messages
# ================== 配置 ==================
OPENAI_API_KEY = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxx"
OPENAI_API_BASE = "https://dashscope.aliyuncs.com/compatible-mode/v1"
# 指向你的 mcp.json 路径（可改为你实际路径）
MCP_JSON_PATH = "/home/user/桌面/langchain-mcp-demo/mcp.json"
class State(TypedDict):
messages: Annotated[list, add_messages]
class MCPAgent:
def __init__(self, backend: str = "ollama"):
print("🚀 初始化 MCPAgent", flush=True)
# ——– 模型选择 ——–
if backend == "ollama":
self.llm = ChatOllama(
model="qwen2.5:7b",
temperature=0.7,
)
print("🔹 使用 Ollama", flush=True)
else:
self.llm = ChatOpenAI(
model="qwen3-30b-a3b-instruct-2507",
temperature=0.7,
api_key=OPENAI_API_KEY,
base_url=OPENAI_API_BASE,
)
print("🔹 使用 OpenAI API", flush=True)
# ——– 从 mcp.json 加载 MCP 配置 ——–
if not os.path.exists(MCP_JSON_PATH):
raise FileNotFoundError(f"❌ 找不到 mcp.json: {MCP_JSON_PATH}")
with open(MCP_JSON_PATH, "r", encoding="utf-8") as f:
mcp_config = json.load(f)
servers = mcp_config.get("mcpServers", {})
if not servers:
raise ValueError("❌ mcp.json 中没有 mcpServers 配置")
# 构建 MultiServerMCPClient 所需的格式
client_config = {}
for name, server in servers.items():
client_config[name] = {
"transport": "stdio",
"command": server["command"],
"args": server["args"]
}
print("🔄 正在连接你自定义的 MCP 服务器（来自 mcp.json）…", flush=True)
client = MultiServerMCPClient(client_config)
# ⚠️ get_tools 是唯一需要等的地方
self.tools = asyncio.run(client.get_tools())
if not self.tools:
raise RuntimeError("❌ MCP 工具为空，请确认 pymcp.py 是否能正常启动并返回工具")
print(f"✅ 已自动注册 {len(self.tools)} 个 MCP 工具: {[t.name for t in self.tools]}", flush=True)
# ⭐ 核心：直接 bind_tools
self.llm = self.llm.bind_tools(self.tools)
# ——– LangGraph ——–
self.graph = self._build_graph()
self.history = []
def _agent(self, state: State):
return {"messages": [self.llm.invoke(state["messages"])]}
def _build_graph(self):
g = StateGraph(State)
g.add_node("agent", self._agent)
g.add_node("tools", ToolNode(self.tools))
g.add_edge(START, "agent")
g.add_conditional_edges("agent", tools_condition)
g.add_edge("tools", "agent")
return g.compile(checkpointer=MemorySaver())
async def run(self):
print("\\n🤖 MCP Chat Agent 已启动", flush=True)
print("示例：", flush=True)
print(" 今天是几号？", flush=True)
print(" 用中文格式返回当前日期", flush=True)
print("\\n", flush=True)
while True:
user = input("你: ").strip()
if user in ("exit", "quit"):
break
self.history.append(("user", user))
async for event in self.graph.astream_events(
{"messages": self.history},
config={"configurable": {"thread_id": "default"}},
version="v2",
):
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"]
if chunk.content:
print(chunk.content, end="", flush=True)
elif event["event"] == "on_tool_start":
print(f"\\n🛠️ 调用工具：{event['name']}", flush=True)
elif event["event"] == "on_tool_end":
print("\\n✅ 工具执行完成", flush=True)
print()
# ================== 启动 ==================
if __name__ == "__main__":
try:
choice = input("选择模型（1=Ollama，2=OpenAI）：").strip()
backend = "ollama" if choice != "2" else "openai"
agent = MCPAgent(backend=backend)
asyncio.run(agent.run())
except Exception as e:
print(f"\\n❌ 启动失败：{e}", flush=True)
print("\\n排查顺序：", flush=True)
print("1. 确认 mcp.json 路径是否正确", flush=True)
print("2. 手动运行：/usr/bin/python /home/user/桌面/langchain-mcp-demo/pymcp.py", flush=True)
print(" 看是否能正常接收 JSON 并返回 initialize 响应", flush=True)
print("3. 确保 pymcp.py 所在目录有 tools/ 和 util/ 子目录", flush=True)
这个主程序实现了以下功能：
6. 配置文件（mcp.json）
{
"mcpServers": {
"mcp-server-python": {
"command": "/usr/bin/python",
"args": [
"/home/user/桌面/langchain-mcp-demo/pymcp.py"
]
}
}
}
注意：项目还需要安装langchain_ollama、langchain_openai和langchain-mcp-adapters等额外依赖。
用python获取所有的源代码
"""
🚀 自解压源码文件
运行后将还原所有源代码并生成 source_code.zip
"""
from pathlib import Path
import zipfile
import json
FILES = json.loads("{\\"demo.py\\": \\"#!/usr/bin/env python3\\\\n\\\\\\"\\\\\\"\\\\\\"\\\\n\\u6700\\u7ec8\\u7a33\\u5b9a\\u7248 MCP Chat Agent\\uff08\\u652f\\u6301 mcp.json\\uff09\\\\n- \\u81ea\\u52a8\\u52a0\\u8f7d mcp.json \\u914d\\u7f6e\\\\n- \\u542f\\u52a8\\u4f60\\u81ea\\u5df1\\u7684 pymcp.py \\u4f5c\\u4e3a MCP \\u670d\\u52a1\\u5668\\\\n- \\u652f\\u6301 Ollama / OpenAI \\u4e8c\\u9009\\u4e00\\\\n\\\\\\"\\\\\\"\\\\\\"\\\\n\\\\nimport asyncio\\\\nimport json\\\\nimport os\\\\nfrom typing import Annotated, TypedDict\\\\n\\\\nfrom langchain_ollama import ChatOllama\\\\nfrom langchain_openai import ChatOpenAI\\\\nfrom langchain_mcp_adapters.client import MultiServerMCPClient\\\\n\\\\nfrom langgraph.graph import StateGraph, START\\\\nfrom langgraph.prebuilt import ToolNode, tools_condition\\\\nfrom langgraph.checkpoint.memory import MemorySaver\\\\nfrom langgraph.graph.message import add_messages\\\\n\\\\n\\\\n# ================== \\u914d\\u7f6e ==================\\\\nOPENAI_API_KEY = \\\\\\"sk-xxxxxxxxxxxxxxxxxxxxx\\\\\\"\\\\nOPENAI_API_BASE = \\\\\\"https://dashscope.aliyuncs.com/compatible-mode/v1\\\\\\"\\\\n\\\\n# \\u6307\\u5411\\u4f60\\u7684 mcp.json \\u8def\\u5f84\\uff08\\u53ef\\u6539\\u4e3a\\u4f60\\u5b9e\\u9645\\u8def\\u5f84\\uff09\\\\nMCP_JSON_PATH = \\\\\\"/home/user/\\u684c\\u9762/langchain-mcp-demo/mcp.json\\\\\\"\\\\n\\\\n\\\\nclass State(TypedDict):\\\\n messages: Annotated[list, add_messages]\\\\n\\\\n\\\\nclass MCPAgent:\\\\n def __init__(self, backend: str = \\\\\\"ollama\\\\\\"):\\\\n print(\\\\\\"\\ud83d\\ude80 \\u521d\\u59cb\\u5316 MCPAgent\\\\\\", flush=True)\\\\n\\\\n # ——– \\u6a21\\u578b\\u9009\\u62e9 ——–\\\\n if backend == \\\\\\"ollama\\\\\\":\\\\n self.llm = ChatOllama(\\\\n model=\\\\\\"qwen2.5:7b\\\\\\",\\\\n temperature=0.7,\\\\n )\\\\n print(\\\\\\"\\ud83d\\udd39 \\u4f7f\\u7528 Ollama\\\\\\", flush=True)\\\\n else:\\\\n self.llm = ChatOpenAI(\\\\n model=\\\\\\"qwen3-30b-a3b-instruct-2507\\\\\\",\\\\n temperature=0.7,\\\\n api_key=OPENAI_API_KEY,\\\\n base_url=OPENAI_API_BASE,\\\\n )\\\\n print(\\\\\\"\\ud83d\\udd39 \\u4f7f\\u7528 OpenAI API\\\\\\", flush=True)\\\\n\\\\n # ——– \\u4ece mcp.json \\u52a0\\u8f7d MCP \\u914d\\u7f6e ——–\\\\n if not os.path.exists(MCP_JSON_PATH):\\\\n raise FileNotFoundError(f\\\\\\"\\u274c \\u627e\\u4e0d\\u5230 mcp.json: {MCP_JSON_PATH}\\\\\\")\\\\n\\\\n with open(MCP_JSON_PATH, \\\\\\"r\\\\\\", encoding=\\\\\\"utf-8\\\\\\") as f:\\\\n mcp_config = json.load(f)\\\\n\\\\n servers = mcp_config.get(\\\\\\"mcpServers\\\\\\", {})\\\\n if not servers:\\\\n raise ValueError(\\\\\\"\\u274c mcp.json \\u4e2d\\u6ca1\\u6709 mcpServers \\u914d\\u7f6e\\\\\\")\\\\n\\\\n # \\u6784\\u5efa MultiServerMCPClient \\u6240\\u9700\\u7684\\u683c\\u5f0f\\\\n client_config = {}\\\\n for name, server in servers.items():\\\\n client_config[name] = {\\\\n \\\\\\"transport\\\\\\": \\\\\\"stdio\\\\\\",\\\\n \\\\\\"command\\\\\\": server[\\\\\\"command\\\\\\"],\\\\n \\\\\\"args\\\\\\": server[\\\\\\"args\\\\\\"]\\\\n }\\\\n\\\\n print(\\\\\\"\\ud83d\\udd04 \\u6b63\\u5728\\u8fde\\u63a5\\u4f60\\u81ea\\u5b9a\\u4e49\\u7684 MCP \\u670d\\u52a1\\u5668\\uff08\\u6765\\u81ea mcp.json\\uff09…\\\\\\", flush=True)\\\\n client = MultiServerMCPClient(client_config)\\\\n\\\\n # \\u26a0\\ufe0f get_tools \\u662f\\u552f\\u4e00\\u9700\\u8981\\u7b49\\u7684\\u5730\\u65b9\\\\n self.tools = asyncio.run(client.get_tools())\\\\n\\\\n if not self.tools:\\\\n raise RuntimeError(\\\\\\"\\u274c MCP \\u5de5\\u5177\\u4e3a\\u7a7a\\uff0c\\u8bf7\\u786e\\u8ba4 pymcp.py \\u662f\\u5426\\u80fd\\u6b63\\u5e38\\u542f\\u52a8\\u5e76\\u8fd4\\u56de\\u5de5\\u5177\\\\\\")\\\\n\\\\n print(f\\\\\\"\\u2705 \\u5df2\\u81ea\\u52a8\\u6ce8\\u518c {len(self.tools)} \\u4e2a MCP \\u5de5\\u5177: {[t.name for t in self.tools]}\\\\\\", flush=True)\\\\n\\\\n # \\u2b50 \\u6838\\u5fc3\\uff1a\\u76f4\\u63a5 bind_tools\\\\n self.llm = self.llm.bind_tools(self.tools)\\\\n\\\\n # ——– LangGraph ——–\\\\n self.graph = self._build_graph()\\\\n self.history = []\\\\n\\\\n def _agent(self, state: State):\\\\n return {\\\\\\"messages\\\\\\": [self.llm.invoke(state[\\\\\\"messages\\\\\\"])]}\\\\n\\\\n def _build_graph(self):\\\\n g = StateGraph(State)\\\\n g.add_node(\\\\\\"agent\\\\\\", self._agent)\\\\n g.add_node(\\\\\\"tools\\\\\\", ToolNode(self.tools))\\\\n g.add_edge(START, \\\\\\"agent\\\\\\")\\\\n g.add_conditional_edges(\\\\\\"agent\\\\\\", tools_condition)\\\\n g.add_edge(\\\\\\"tools\\\\\\", \\\\\\"agent\\\\\\")\\\\n return g.compile(checkpointer=MemorySaver())\\\\n\\\\n async def run(self):\\\\n print(\\\\\\"\\\\\\\\n\\ud83e\\udd16 MCP Chat Agent \\u5df2\\u542f\\u52a8\\\\\\", flush=True)\\\\n print(\\\\\\"\\u793a\\u4f8b\\uff1a\\\\\\", flush=True)\\\\n print(\\\\\\" \\u4eca\\u5929\\u662f\\u51e0\\u53f7\\uff1f\\\\\\", flush=True)\\\\n print(\\\\\\" \\u7528\\u4e2d\\u6587\\u683c\\u5f0f\\u8fd4\\u56de\\u5f53\\u524d\\u65e5\\u671f\\\\\\", flush=True)\\\\n print(\\\\\\"\\\\\\\\n\\\\\\", flush=True)\\\\n\\\\n while True:\\\\n user = input(\\\\\\"\\u4f60: \\\\\\").strip()\\\\n if user in (\\\\\\"exit\\\\\\", \\\\\\"quit\\\\\\"):\\\\n break\\\\n\\\\n self.history.append((\\\\\\"user\\\\\\", user))\\\\n\\\\n async for event in self.graph.astream_events(\\\\n {\\\\\\"messages\\\\\\": self.history},\\\\n config={\\\\\\"configurable\\\\\\": {\\\\\\"thread_id\\\\\\": \\\\\\"default\\\\\\"}},\\\\n version=\\\\\\"v2\\\\\\",\\\\n ):\\\\n if event[\\\\\\"event\\\\\\"] == \\\\\\"on_chat_model_stream\\\\\\":\\\\n chunk = event[\\\\\\"data\\\\\\"][\\\\\\"chunk\\\\\\"]\\\\n if chunk.content:\\\\n print(chunk.content, end=\\\\\\"\\\\\\", flush=True)\\\\n\\\\n elif event[\\\\\\"event\\\\\\"] == \\\\\\"on_tool_start\\\\\\":\\\\n print(f\\\\\\"\\\\\\\\n\\ud83d\\udee0\\ufe0f \\u8c03\\u7528\\u5de5\\u5177\\uff1a{event['name']}\\\\\\", flush=True)\\\\n\\\\n elif event[\\\\\\"event\\\\\\"] == \\\\\\"on_tool_end\\\\\\":\\\\n print(\\\\\\"\\\\\\\\n\\u2705 \\u5de5\\u5177\\u6267\\u884c\\u5b8c\\u6210\\\\\\", flush=True)\\\\n\\\\n print()\\\\n\\\\n\\\\n# ================== \\u542f\\u52a8 ==================\\\\nif __name__ == \\\\\\"__main__\\\\\\":\\\\n try:\\\\n choice = input(\\\\\\"\\u9009\\u62e9\\u6a21\\u578b\\uff081=Ollama\\uff0c2=OpenAI\\uff09\\uff1a\\\\\\").strip()\\\\n backend = \\\\\\"ollama\\\\\\" if choice != \\\\\\"2\\\\\\" else \\\\\\"openai\\\\\\"\\\\n\\\\n agent = MCPAgent(backend=backend)\\\\n asyncio.run(agent.run())\\\\n\\\\n except Exception as e:\\\\n print(f\\\\\\"\\\\\\\\n\\u274c \\u542f\\u52a8\\u5931\\u8d25\\uff1a{e}\\\\\\", flush=True)\\\\n print(\\\\\\"\\\\\\\\n\\u6392\\u67e5\\u987a\\u5e8f\\uff1a\\\\\\", flush=True)\\\\n print(\\\\\\"1. \\u786e\\u8ba4 mcp.json \\u8def\\u5f84\\u662f\\u5426\\u6b63\\u786e\\\\\\", flush=True)\\\\n print(\\\\\\"2. \\u624b\\u52a8\\u8fd0\\u884c\\uff1a/usr/bin/python /home/user/\\u684c\\u9762/langchain-mcp-demo/pymcp.py\\\\\\", flush=True)\\\\n print(\\\\\\" \\u770b\\u662f\\u5426\\u80fd\\u6b63\\u5e38\\u63a5\\u6536 JSON \\u5e76\\u8fd4\\u56de initialize \\u54cd\\u5e94\\\\\\", flush=True)\\\\n print(\\\\\\"3. \\u786e\\u4fdd pymcp.py \\u6240\\u5728\\u76ee\\u5f55\\u6709 tools/ \\u548c util/ \\u5b50\\u76ee\\u5f55\\\\\\", flush=True)\\", \\"mcp.json\\": \\"{\\\\n \\\\\\"mcpServers\\\\\\": {\\\\n \\\\\\"mcp-server-python\\\\\\": {\\\\n \\\\\\"command\\\\\\": \\\\\\"/usr/bin/python\\\\\\",\\\\n \\\\\\"args\\\\\\": [\\\\n \\\\\\"/home/user/\\u684c\\u9762/langchain-mcp-demo/pymcp.py\\\\\\"\\\\n ]\\\\n }\\\\n }\\\\n}\\", \\"mcp_api.py\\": \\"# mcp_api.py\\\\n\\\\nfrom tools.get_date import get_date\\\\nfrom tools.time_tool import get_current_time \\\\nimport json\\\\nimport sys\\\\n# \\u53ea\\u6ce8\\u518c\\u4e00\\u4e2a\\u5de5\\u5177\\uff1a\\u83b7\\u53d6\\u5f53\\u524d\\u65e5\\u671f\\\\nTOOLS = {\\\\n \\\\\\"get_current_time\\\\\\": {\\\\n \\\\\\"name\\\\\\": \\\\\\"get_current_time\\\\\\",\\\\n \\\\\\"description\\\\\\": \\\\\\"\\u83b7\\u53d6\\u5f53\\u524d\\u65f6\\u95f4\\uff0c\\u652f\\u6301\\u81ea\\u5b9a\\u4e49\\u683c\\u5f0f\\u6216 ISO 8601 \\u6807\\u51c6\\u683c\\u5f0f\\\\\\",\\\\n \\\\\\"inputSchema\\\\\\": {\\\\n \\\\\\"type\\\\\\": \\\\\\"object\\\\\\",\\\\n \\\\\\"properties\\\\\\": {\\\\n \\\\\\"format\\\\\\": {\\\\n \\\\\\"type\\\\\\": \\\\\\"string\\\\\\",\\\\n \\\\\\"description\\\\\\": \\\\\\"\\u65f6\\u95f4\\u683c\\u5f0f\\uff0c\\u4f8b\\u5982 '%H:%M'\\u3001'%H\\u65f6%M\\u5206'\\uff0c\\u6216 'iso' \\u8868\\u793a ISO 8601\\\\\\",\\\\n \\\\\\"default\\\\\\": \\\\\\"%H:%M:%S\\\\\\"\\\\n }\\\\n },\\\\n \\\\\\"required\\\\\\": []\\\\n },\\\\n \\\\\\"func\\\\\\": get_current_time\\\\n },\\\\n \\\\\\"get_date\\\\\\": {\\\\n \\\\\\"name\\\\\\": \\\\\\"get_date\\\\\\",\\\\n \\\\\\"description\\\\\\": \\\\\\"\\u83b7\\u53d6\\u5f53\\u524d\\u65e5\\u671f\\uff0c\\u652f\\u6301\\u81ea\\u5b9a\\u4e49\\u683c\\u5f0f\\uff08\\u5982 '%Y\\u5e74%m\\u6708%d\\u65e5'\\uff09\\\\\\",\\\\n \\\\\\"inputSchema\\\\\\": {\\\\n \\\\\\"type\\\\\\": \\\\\\"object\\\\\\",\\\\n \\\\\\"properties\\\\\\": {\\\\n \\\\\\"format\\\\\\": {\\\\n \\\\\\"type\\\\\\": \\\\\\"string\\\\\\",\\\\n \\\\\\"description\\\\\\": \\\\\\"\\u65e5\\u671f\\u683c\\u5f0f\\uff0c\\u4f8b\\u5982 '%Y-%m-%d'\\u3001'%Y\\u5e74%m\\u6708%d\\u65e5' \\u7b49\\\\\\",\\\\n \\\\\\"default\\\\\\": \\\\\\"%Y-%m-%d\\\\\\"\\\\n }\\\\n },\\\\n \\\\\\"required\\\\\\": []\\\\n },\\\\n \\\\\\"func\\\\\\": get_date\\\\n }\\\\n}\\\\n\\\\ndef send_response(req_id, result=None, error=None):\\\\n \\\\\\"\\\\\\"\\\\\\"\\\\n \\u53d1\\u9001\\u7b26\\u5408 JSON-RPC 2.0 \\u6807\\u51c6\\u7684\\u54cd\\u5e94\\\\n \\\\\\"\\\\\\"\\\\\\"\\\\n response = {\\\\n \\\\\\"jsonrpc\\\\\\": \\\\\\"2.0\\\\\\", # \\u2b05\\ufe0f \\u5173\\u952e\\uff01\\u5fc5\\u987b\\u52a0\\u4e0a\\\\n \\\\\\"id\\\\\\": req_id\\\\n }\\\\n if error is not None:\\\\n response[\\\\\\"error\\\\\\"] = {\\\\n \\\\\\"code\\\\\\": -32000, # \\u81ea\\u5b9a\\u4e49\\u9519\\u8bef\\u7801\\\\n \\\\\\"message\\\\\\": str(error)\\\\n }\\\\n else:\\\\n response[\\\\\\"result\\\\\\"] = result\\\\n\\\\n # \\u8f93\\u51fa\\u5230 stdout\\uff08MCP \\u534f\\u8bae\\u8981\\u6c42\\uff09\\\\n print(json.dumps(response, ensure_ascii=False, separators=(',', ':')))\\\\n sys.stdout.flush()\\", \\"pymcp.py\\": \\"# pymcp.py\\\\n# \\u4e3b MCP \\u670d\\u52a1\\u5668\\u5165\\u53e3\\uff1a\\u4ece\\u6807\\u51c6\\u8f93\\u5165\\u8bfb\\u53d6\\u8bf7\\u6c42\\uff0c\\u8c03\\u7528\\u5de5\\u5177\\u5e76\\u8fd4\\u56de\\u7ed3\\u679c\\\\n\\\\nfrom mcp_api import TOOLS, send_response\\\\nimport sys\\\\nimport json\\\\nfrom util.logger import log # \\u65e5\\u5fd7\\u8bb0\\u5f55\\u5de5\\u5177\\\\n\\\\n\\\\ndef run_server():\\\\n \\\\\\"\\\\\\"\\\\\\"\\u8fd0\\u884c MCP \\u670d\\u52a1\\u5668\\u4e3b\\u5faa\\u73af\\\\\\"\\\\\\"\\\\\\"\\\\n for line in sys.stdin:\\\\n # \\u8df3\\u8fc7\\u7a7a\\u884c\\\\n if not line.strip():\\\\n continue\\\\n\\\\n log(f\\\\\\"\\u6536\\u5230\\u8bf7\\u6c42: {line.strip()}\\\\\\")\\\\n\\\\n # \\u89e3\\u6790 JSON \\u8bf7\\u6c42\\\\n try:\\\\n req = json.loads(line)\\\\n except Exception as e:\\\\n log(f\\\\\\"JSON \\u683c\\u5f0f\\u9519\\u8bef: {e}\\\\\\")\\\\n send_response(None, error=f\\\\\\"\\u65e0\\u6548\\u7684 JSON: {e}\\\\\\")\\\\n continue\\\\n\\\\n req_id = req.get(\\\\\\"id\\\\\\")\\\\n method = req.get(\\\\\\"method\\\\\\")\\\\n params = req.get(\\\\\\"params\\\\\\", {})\\\\n\\\\n # \\u521d\\u59cb\\u5316\\u8fde\\u63a5\\uff08\\u5ba2\\u6237\\u7aef\\u9996\\u6b21\\u63e1\\u624b\\uff09\\\\n if method == \\\\\\"initialize\\\\\\":\\\\n # \\u6784\\u5efa\\u5de5\\u5177\\u80fd\\u529b\\u5bf9\\u8c61\\uff08LM Studio \\u8981\\u6c42\\u683c\\u5f0f\\u4e3a\\u5bf9\\u8c61\\uff09\\\\n tools_obj = {}\\\\n for key, meta in TOOLS.items():\\\\n tools_obj[meta[\\\\\\"name\\\\\\"]] = {\\\\n \\\\\\"description\\\\\\": meta[\\\\\\"description\\\\\\"],\\\\n \\\\\\"inputSchema\\\\\\": meta[\\\\\\"inputSchema\\\\\\"]\\\\n }\\\\n\\\\n send_response(req_id, result={\\\\n \\\\\\"protocolVersion\\\\\\": \\\\\\"2025-06-18\\\\\\",\\\\n \\\\\\"serverInfo\\\\\\": {\\\\n \\\\\\"name\\\\\\": \\\\\\"mcp-server-python\\\\\\",\\\\n \\\\\\"version\\\\\\": \\\\\\"0.1.0\\\\\\"\\\\n },\\\\n \\\\\\"capabilities\\\\\\": {\\\\n \\\\\\"tools\\\\\\": tools_obj\\\\n }\\\\n })\\\\n continue\\\\n\\\\n # \\u8c03\\u7528\\u5de5\\u5177\\uff08MCP \\u534f\\u8bae\\u65b9\\u6cd5\\uff1atools/call\\uff09\\\\n if method == \\\\\\"tools/call\\\\\\":\\\\n tool_name = params.get(\\\\\\"name\\\\\\")\\\\n arguments = params.get(\\\\\\"arguments\\\\\\", {})\\\\n\\\\n if tool_name not in TOOLS:\\\\n log(f\\\\\\"\\u672a\\u77e5\\u5de5\\u5177: {tool_name}\\\\\\")\\\\n send_response(req_id, error=f\\\\\\"\\u672a\\u77e5\\u5de5\\u5177: {tool_name}\\\\\\")\\\\n continue\\\\n\\\\n try:\\\\n log(f\\\\\\"\\u6b63\\u5728\\u8c03\\u7528\\u5de5\\u5177 {tool_name}\\uff0c\\u53c2\\u6570: {arguments}\\\\\\")\\\\n result = TOOLS[tool_name][\\\\\\"func\\\\\\"](**arguments)\\\\n log(f\\\\\\"\\u5de5\\u5177 {tool_name} \\u8fd4\\u56de: {result}\\\\\\")\\\\n\\\\n # \\u6309\\u7167 MCP \\u534f\\u8bae\\u8fd4\\u56de content \\u6570\\u7ec4\\\\n send_response(req_id, result={\\\\n \\\\\\"content\\\\\\": [\\\\n {\\\\n \\\\\\"type\\\\\\": \\\\\\"text\\\\\\",\\\\n \\\\\\"text\\\\\\": str(result)\\\\n }\\\\n ]\\\\n })\\\\n except Exception as e:\\\\n log(f\\\\\\"\\u5de5\\u5177 {tool_name} \\u6267\\u884c\\u51fa\\u9519: {e}\\\\\\")\\\\n send_response(req_id, error=str(e))\\\\n continue\\\\n\\\\n # \\u5ba2\\u6237\\u7aef\\u521d\\u59cb\\u5316\\u5b8c\\u6210\\u901a\\u77e5\\uff08\\u53ef\\u5ffd\\u7565\\uff09\\\\n if method == \\\\\\"notifications/initialized\\\\\\":\\\\n # \\u8fd9\\u662f\\u5ba2\\u6237\\u7aef\\u53d1\\u6765\\u7684\\u901a\\u77e5\\uff0c\\u670d\\u52a1\\u5668\\u65e0\\u9700\\u56de\\u590d\\uff01\\\\n log(\\\\\\"\\u6536\\u5230 initialized \\u901a\\u77e5\\uff0c\\u5ffd\\u7565\\\\\\")\\\\n continue # \\u4e0d\\u8981 send_response\\uff01\\\\n\\\\n # \\u5217\\u51fa\\u6240\\u6709\\u5de5\\u5177\\uff08\\u517c\\u5bb9\\u65e7\\u534f\\u8bae\\uff09\\\\n if method == \\\\\\"tools/list\\\\\\":\\\\n tools_info = []\\\\n for key, meta in TOOLS.items():\\\\n tools_info.append({\\\\n \\\\\\"name\\\\\\": meta[\\\\\\"name\\\\\\"],\\\\n \\\\\\"description\\\\\\": meta[\\\\\\"description\\\\\\"],\\\\n \\\\\\"inputSchema\\\\\\": meta[\\\\\\"inputSchema\\\\\\"]\\\\n })\\\\n send_response(req_id, result={\\\\\\"tools\\\\\\": tools_info})\\\\n continue\\\\n\\\\n # \\u5ffd\\u7565\\u5176\\u4ed6\\u901a\\u77e5\\u7c7b\\u6d88\\u606f\\uff08\\u5982 notifications/progress \\u7b49\\uff09\\\\n if method and method.startswith(\\\\\\"notifications/\\\\\\"):\\\\n # \\u901a\\u77e5\\u7c7b\\u6d88\\u606f\\u4e0d\\u9700\\u8981\\u54cd\\u5e94\\uff0c\\u4f46\\u4e3a\\u5b89\\u5168\\u8d77\\u89c1\\u8fd4\\u56de\\u4e00\\u4e2a\\u7a7a\\u786e\\u8ba4\\\\n send_response(None, result={\\\\\\"status\\\\\\": \\\\\\"ok\\\\\\"})\\\\n continue\\\\n\\\\n # \\u5176\\u4ed6\\u672a\\u652f\\u6301\\u7684\\u65b9\\u6cd5\\\\n log(f\\\\\\"\\u6536\\u5230\\u672a\\u77e5\\u65b9\\u6cd5: {method}\\\\\\")\\\\n send_response(req_id, error=f\\\\\\"\\u672a\\u77e5\\u65b9\\u6cd5: {method}\\\\\\")\\\\n\\\\n\\\\nif __name__ == \\\\\\"__main__\\\\\\":\\\\n run_server()\\", \\"util/logger.py\\": \\"# util/logger.py\\\\n# \\u7b80\\u5355\\u65e5\\u5fd7\\u5de5\\u5177\\uff1a\\u5c06\\u65e5\\u5fd7\\u6253\\u5370\\u5230 stderr\\uff0c\\u907f\\u514d\\u5e72\\u6270 MCP \\u7684 stdout \\u901a\\u4fe1\\\\n\\\\nimport sys\\\\nfrom datetime import datetime\\\\n\\\\n\\\\ndef log(message: str):\\\\n \\\\\\"\\\\\\"\\\\\\"\\\\n \\u8bb0\\u5f55\\u65e5\\u5fd7\\u5230\\u6807\\u51c6\\u9519\\u8bef\\uff08stderr\\uff09\\uff0c\\u5e26\\u65f6\\u95f4\\u6233\\u3002\\\\n \\\\n \\u53c2\\u6570:\\\\n message (str): \\u8981\\u8bb0\\u5f55\\u7684\\u65e5\\u5fd7\\u5185\\u5bb9\\\\n \\\\\\"\\\\\\"\\\\\\"\\\\n timestamp = datetime.now().strftime(\\\\\\"%Y-%m-%d %H:%M:%S\\\\\\")\\\\n print(f\\\\\\"[{timestamp}] {message}\\\\\\", file=sys.stderr, flush=True)\\", \\"tools/get_date.py\\": \\"# tools/get_date.py\\\\n\\\\nfrom datetime import datetime\\\\n\\\\ndef get_date(format: str = \\\\\\"%Y-%m-%d\\\\\\") -> str:\\\\n \\\\\\"\\\\\\"\\\\\\"\\\\n Returns the current date as a string.\\\\n \\\\n Args:\\\\n format (str): Date format string (default: \\\\\\"%Y-%m-%d\\\\\\")\\\\n \\\\n Returns:\\\\n str: Formatted current date\\\\n \\\\\\"\\\\\\"\\\\\\"\\\\n return datetime.now().strftime(format)\\", \\"tools/time_tool.py\\": \\"# tools/time_tool.py\\\\nfrom datetime import datetime\\\\n\\\\ndef get_current_time(format: str = \\\\\\"%H:%M:%S\\\\\\") -> str:\\\\n \\\\\\"\\\\\\"\\\\\\"\\\\n \\u83b7\\u53d6\\u5f53\\u524d\\u65f6\\u95f4\\uff0c\\u652f\\u6301\\u81ea\\u5b9a\\u4e49\\u683c\\u5f0f\\u3002\\\\n \\\\n Args:\\\\n format (str): \\u65f6\\u95f4\\u683c\\u5f0f\\u5b57\\u7b26\\u4e32\\uff0c\\u9ed8\\u8ba4\\u4e3a \\\\\\"%H:%M:%S\\\\\\"\\\\n \\u793a\\u4f8b\\uff1a\\\\n – \\\\\\"%H:%M\\\\\\" \\u2192 \\\\\\"14:30\\\\\\"\\\\n – \\\\\\"%H\\u65f6%M\\u5206%S\\u79d2\\\\\\" \\u2192 \\\\\\"14\\u65f630\\u520645\\u79d2\\\\\\"\\\\n – \\\\\\"iso\\\\\\" \\u2192 ISO 8601 \\u683c\\u5f0f\\uff08\\u5982 \\\\\\"2026-01-16T14:30:45.123456\\\\\\"\\uff09\\\\n \\\\n Returns:\\\\n str: \\u683c\\u5f0f\\u5316\\u540e\\u7684\\u65f6\\u95f4\\u5b57\\u7b26\\u4e32\\\\n \\\\\\"\\\\\\"\\\\\\"\\\\n if format == \\\\\\"iso\\\\\\":\\\\n return datetime.now().isoformat()\\\\n else:\\\\n return datetime.now().strftime(format)\\", \\"tools/__init__.py\\": \\"\\"}")
BASE_DIR = Path("extracted_source")
ZIP_NAME = "source_code.zip"
def main():
BASE_DIR.mkdir(exist_ok=True)
# 写回所有文件
for path, content in FILES.items():
file_path = BASE_DIR / path
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(content, encoding="utf-8", errors="replace")
# 打包为 zip
with zipfile.ZipFile(ZIP_NAME, "w", zipfile.ZIP_DEFLATED) as zf:
for file in BASE_DIR.rglob("*"):
if file.is_file():
zf.write(file, arcname=file.relative_to(BASE_DIR))
print("✅ 源代码已还原并压缩为", ZIP_NAME)
if __name__ == "__main__":
main()
python aa.py
├── extracted_source
│ ├── demo.py
│ ├── mcp_api.py
│ ├── mcp.json
│ ├── pymcp.py
│ ├── tools
│ │ ├── get_date.py
│ │ ├── __init__.py
│ │ └── time_tool.py
│ └── util
│ └── logger.py
├── self_extract.py
└── source_code.zip
运行说明
总结
这个项目展示了如何使用LangChain和MCP协议构建一个强大的智能对话系统。通过MCP协议，我们可以轻松地将自定义工具集成到AI模型中，让模型能够执行特定任务。这种架构具有很好的扩展性，可以方便地添加更多工具和服务。
创作不易，记得点赞、收藏、加关注！
