MCP(模型上下文协议)是AI工具和资源的一项新兴标准。该标准与普通的REST API服务器兼容,但增加了额外的元数据,以机器可读的方式描述工具、资源和提示。这为我们提供了一个很好的机会,可以创建完全自动映射到这些MCP服务器的Python模块。这种方法的最大优势在于我们可以像使用原生Python库一样使用任何MCP服务器,无需任何配置。这对于创建映射到REST API的Python软件开发套件来说非常有用,因为这是一个极其常见且相当手动的过程。现在,如果托管REST API的组织也提供了MCP接口,我们就可以毫不费力地自动生成一个Python SDK!如果你对这一切还不完全清楚,没关系。即使不了解MCP的所有细节,你仍然可以利用mcp2py的强大功能。你需要知道的是:如果你想编程地与网站交互,它们很可能有一个API,并且随着时间的推移,它们很可能会为这个API提供一个MCP接口。如果他们确实提供了这样的接口,你就无需学习一整套网络编程技能,只需使用mcp2py加载MCP服务器并立即开始调用函数,就像在使用原生Python库一样!
另一个值得注意的酷特性是服务器不必远程运行。你现在可以在自己的个人电脑上运行很多服务器。这对于让不同程序(可能使用不同的编程语言)互相通信非常有用。随着你安装的应用越来越多,它们会在你的机器上打开一个小的本地服务器,以便大型语言模型(LLMs)能够与其互动。这时你可以利用mcp2py与这些本地服务器进行互动。例如,Slack可以打开一个服务器让你查询消息。如果是这样,你可以使用mcp2py并拥有一个Python模块(本质上是一个库),让你可以直接从Python查询Slack消息。非常强大!
这里有一个非常简单的例子,展示如何使用mcp2py与本地文件系统交互。这并不是非常有用,因为你完全可以使用内置的Python库来实现相同的功能,但它作为一个非常简单的例子,展示了mcp2py的工作原理。在这段代码中,我们使用load来启动MCP服务器(在这种情况下,它是一个Node.js服务器)并连接到它。一旦连接成功,我们就可以像调用原生Python函数一样调用list_directory工具:
from mcp2py import load
fstools = load("npx -y @modelcontextprotocol/server-filesystem /home")
fstools.list_directory("/home")
[DIR] maxime
这类似于使用Python中的os库:
import os
os.listdir("/home")
['maxime']
主要的区别在于,我们不是直接从Python到系统,而是向本地Node(JavaScript)服务器发送命令,而该服务器具有一些“安全”功能。例如,我们不允许搜索/home之外的内容,因为我们已经将其设置为根目录。当你要将文件系统暴露给LLM时,这些功能非常有用。
1. 安装
你可以通过pip安装mcp2py:
pip install mcp2py
Python长期以来一直存在没有标准方式管理依赖项的问题。为了避免依赖冲突,建议使用虚拟环境。我最喜欢的方法是使用uv(参见这里:
https://docs.astral.sh/uv/getting-started/installation/)。然后你可以创建一个新的环境并安装mcp2py,如下所示:
# 如果还没有安装uv,请先安装
curl -LsSf https://astral.sh/uv/install.sh | sh
# 创建一个新的项目并带有虚拟环境
uv init my-mcp-project
cd my-mcp-project
# 安装mcp2py
uv add mcp2py
# 激活环境并开始编码
uv run python
2. 使用它
from mcp2py import load
# 加载任何具有OAuth身份验证的MCP服务器
notion = load("https://mcp.notion.com/mcp", auth="oauth")
# 浏览器会自动打开进行OAuth登录
# 认证后,你可以使用工具
notion.notion_get_self()
3. 就这么简单!
服务器作为子进程运行,工具是Python方法,一切都正常工作。
MCP服务器通过协议公开工具、资源和提示。mcp2py将它们转换为{python}:
即插即用™ - 但你可以自定义一切
mcp2py旨在为研究人员、数据分析员和{python}初学者提供一种尝试MCP服务器而不必处理复杂性的方法。同时,它为构建生产应用程序的开发者提供了完全控制。
默认零配置:
高级用户的无上限:
**你的{python} REPL/代码成为MCP客户端。**服务器是一个独立的进程(Node.js、{python}或其他),mcp2py通过JSON-RPC与之通信。你的{python}代码可以:
from mcp2py import load
# 加载任何MCP服务器 - 就这么简单!
server = load("https://api.example.com/mcp")
# 如果需要登录:
# → 浏览器自动打开
# → 你登录一次
# → 浏览器关闭
# → 完成!
# 如果需要你的输入:
# → 出现友好的终端提示
# → 你回答
# → 代码继续执行!
# 如果需要AI帮助(采样):
# → 使用你的ANTHROPIC_API_KEY或OPENAI_API_KEY
# → 自动处理
# → 你甚至不会注意到!
# 直接使用工具!
result = server.analyze_data(dataset="sales_2024.csv")
print(result)
就这么简单。无需配置。无需设置。即插即用。
from mcp2py import load
# 加载MCP服务器 - 简单明了
weather = load("npx -y @h1deya/mcp-server-weather")
# 或从远程HTTP服务器(SSE/HTTP流传输)
api = load("https://api.example.com/mcp")
# 带有身份验证
api = load("https://api.example.com/mcp", headers={"Authorization": "Bearer YOUR_TOKEN"})
# 或从{python}脚本
travel = load("{python} my_mcp_server.py")
# 工具变成函数
alerts = weather.get_alerts(state="CA")
forecast = weather.get_forecast(latitude=37.7749, longitude=-122.4194)
print(forecast)
# 资源变成属性
print(weather.API_DOCUMENTATION) # 常量资源
print(weather.current_config) # 动态资源
# 提示变成模板函数
prompt = weather.create_weather_report(location="NYC", style="casual")
.tools 属性提供了一组可调用的{python}函数:
from mcp2py import load
server = load("npx -y @modelcontextprotocol/server-filesystem /tmp")
# 获取可调用的函数
tools = server.tools
# [<function read_file>, <function write_file>, ...]
# 每个函数都有 __name__ 和 __doc__
print(tools[0].__name__) # "read_file"
print(tools[0].__doc__) # "从文件系统读取文件"
# 并且它们是可调用的!
result = tools[0](path="/tmp/test.txt")
.tools 属性为你提供了可用于DSPy和Claudette等框架的可调用函数:
from mcp2py import load
import dspy
# 加载MCP服务器
travel = load("{python} airline_server.py")
# 与DSPy配合使用 - 直接传递可调用函数
class CustomerService(dspy.Signature):
user_request: str = dspy.InputField()
result: str = dspy.OutputField()
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
# 直接将工具传递给DSPy(它期望可调用函数)
react = dspy.ReAct(CustomerService, tools=travel.tools)
result = react(user_request="预订2025年9月1日从SFO到JFK的航班")
print(result)
# 也可以与Claudette配合使用
from mcp2py import load
from claudette import Chat
weather = load("npx -y @h1deya/mcp-server-weather")
# Claudette期望可调用函数
chat = Chat(model="claude-3-5-sonnet-20241022", tools=weather.tools)
response = chat("东京天气怎么样?")
# Claudette会根据需要自动调用工具
print(response)
**注意:**对于具有原生MCP支持的SDK(如Anthropic、OpenAI、Google Gemini),请直接使用其内置的MCP集成。.tools 属性适用于期望{python}可调用函数的框架,如DSPy和Claudette。
自动生成存根以获得完美的自动完成功能:
from mcp2py import load
# 存根自动生成到 ~/.cache/mcp2py/stubs/
server = load("npx my-server")
# IDE现在具有完整的自动完成功能和类型提示!
server.search_files(
pattern="*.py", # 类型:str - IDE知道这一点!
max_results=10 # 类型:int,可选 - IDE建议这一点!
) # 返回:dict[str, Any] - IDE显示返回类型!
手动生成存根:
# 将存根生成到特定位置用于你的项目
server = load("npx weather-server")
server.generate_stubs("./stubs/weather.pyi")
# 或让它自动缓存(默认行为)
# 存根保存到:~/.cache/mcp2py/stubs/<command_hash>.pyi
工作原理:
load() 返回一个动态类型的类,所有方法都预先定义好了.pyi存根文件到~/.cache/mcp2py/stubs/供参考无需配置 - 自动完成功能就绪!✨
当你{python}代码充当MCP客户端时,服务器可能会请求以下能力:
当服务器需要LLM完成时,mcp2py会自动处理。
默认:开箱即用
from mcp2py import load
# 直接使用!使用默认的LLM
server = load("npx travel-server")
# 如果服务器需要LLM的帮助,mcp2py:
# 1. 检查环境中的ANTHROPIC_API_KEY或OPENAI_API_KEY
# 2. 自动调用LLM
# 3. 将结果返回给服务器
# 4. 你的代码继续执行!
result = server.book_flight(destination="东京")
配置你首选的LLM:
# 通过环境变量设置(推荐)
import os
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
# 或使用LiteLLM模型字符串全局配置
from mcp2py import configure
configure(
model="claude-3-5-sonnet-20241022" # 或 "gpt-4o", "gemini/gemini-pro" 等
)
# LiteLLM会根据模型名称自动检测正确的API
# 使用标准环境变量:ANTHROPIC_API_KEY, OPENAI_API_KEY �[...]
# 等。
# 现在所有服务器都会使用此LLM进行采样
server = load("npx travel-server")
高级:自定义采样处理器
from mcp2py import load
def my_sampling_handler(messages, model_prefs, system_prompt, max_tokens):
"""对LLM调用的完全控制。"""
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
messages=messages,
max_tokens=max_tokens
)
return response.content[0].text
server = load(
"npx travel-server",
on_sampling=my_sampling_handler # 覆盖默认值
)
禁用采样(为了安全/成本控制):
server = load(
"npx travel-server",
allow_sampling=False # 如果服务器请求LLM,则引发错误
)
当服务器需要用户输入时,mcp2py会自动提示。
默认:终端提示
from mcp2py import load
# 直接使用!终端提示会自动出现
server = load("npx travel-server")
# 服务器询问:"确认$500的预订?"
# 终端显示:
#
# 服务器询问:确认$500的预订?
# confirm_booking (布尔值):y/n
#
# 你输入:y
# 代码继续执行!
result = server.book_flight(destination="巴黎")
你会看到:
正在调用book_flight...
┌─────────────────────────────────────────┐
│ 🔔 服务器需要你的输入 │
├─────────────────────────────────────────┤
│ 确认$500的预订? │
│ │
│ confirm_booking (布尔值):y/n │
│ seat_preference (靠窗/过道/中间): │
│ meal_preference (可选): │
└─────────────────────────────────────────┘
> y
> 窗户
> 素食
预订已确认!
高级:自定义引诱处理器
from mcp2py import load
def my_input_handler(message, schema):
"""用户输入的自定义UI。"""
# 构建GUI、网页表单、语音输入等。
from tkinter import simpledialog
return simpledialog.askstring("服务器请求", message)
server = load(
"npx travel-server",
on_elicitation=my_input_handler
)
禁用引诱(用于自动化脚本):
server = load(
"npx travel-server",
allow_elicitation=False # 如果服务器请求输入,则引发错误
)
# 或提供预填的答案
server = load(
"npx travel-server",
elicitation_defaults={
"confirm_booking": True,
"seat_preference": "窗户"
}
)
服务器可以询问应关注哪些目录。可选,简单:
# 单个目录
server = load("npx filesystem-server", roots="/home/user/projects")
# 多个目录
server = load(
"npx filesystem-server",
roots=["/home/user/projects", "/tmp/workspace"]
)
# 动态更新根目录
server.set_roots(["/home/user/new-project"])
MCP工具映射到{python}函数,完全支持:
inputSchema生成description 构建dict[str, Any](MCP工具返回JSON)命名约定:蛇形命名(MCP getWeather → {python} get_weather)
# MCP工具定义:
# {
# "name": "searchFiles",
# "description": "搜索匹配模式的文件",
# "inputSchema": {
# "type": "object",
# "properties": {
# "pattern": {"type": "string", "description": "通配符模式"},
# "maxResults": {"type": "integer", "default": 100}
# },
# "required": ["pattern"]
# }
# }
# 生成的{python}:
def search_files(pattern: str, max_results: int = 100) -> dict[str, Any]:
"""搜索匹配模式的文件。
参数:
pattern: 通配符模式
max_results: 最多返回的结果数(默认:100)
"""
...
资源根据其性质映射不同:
# 静态资源(缓存)
API_DOCS: str = server._get_resource("api://docs")
# 动态资源(访问时获取)
@property
def current_status() -> dict[str, Any]:
"""当前服务器状态。"""
return server._get_resource("status://current")
命名约定: