mcp2py：将任何MCP服务器转换为python模块
MCP（模型上下文协议）是人工智能工具和 资源。该标准与普通的REST API服务器兼容，但是 添加额外的元数据来描述工具、资源和提示 机器可读的方式。这为我们提供了一个很好的机会 创建完全自动映射到这些MCP的Python模块 服务器。这种方法的最大优点是我们可以使用任何 MCP服务器，就像它是一个原生Python库，没有 配置。这对于创建Python软件来说可能是一件大事 映射到REST API的开发工具包非常常见 手动过程。现在，如果承载REST API的组织也 提供了一个MCP接口，我们可以自动生成Python SDK 无需付出任何努力！如果你还不完全清楚，不要担心。 您仍然可以在不了解所有信息的情况下利用mcp2py的强大功能 MCP的详细信息。你需要知道的是：如果你想以编程方式 与网站互动，很可能他们有API，随着时间的推移 继续下去，他们很可能有一个用于API的MCP接口。 如果他们这样做了，你就不必学习一整套网络编程 技能，您只需使用mcp2py加载MCP服务器并开始调用 立即执行函数，就像它是一个原生Python库一样！
另一件很酷的事情是，服务器不必运行 远程。你可以（并且有）很多服务器自己运行 现在的个人电脑。这对于具有不同的程序是有用的， 可能使用不同的编程语言相互通信。如 您安装的应用程序将越来越多地打开一个小型本地服务器 如果您的机器允许LLM与它们交互，您还可以 利用mcp2py与这些本地服务器进行交互。这可能看起来 就像Slack打开一个服务器，让你查询你的消息。如果是这样， 然后，您可以使用mcp2py并拥有一个Python模块（ essence），允许您直接从Python查询Slack消息。 超级强大！
概述
下面是一个使用mcp2py与您的 本地文件系统。这不是很有用，因为你可以直接使用 内置的Python库可以做到这一点，但它非常简单 示例说明mcp2py的工作原理。在这段代码中，我们使用 加载以启动MCP服务器（此处为Node.js服务器） case）并连接到它。连接后，我们可以调用list_directory 工具，就像它是一个原生Python函数一样：
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转到 系统，我们向本地节点（JavaScript）服务器发送命令 服务器具有一些“安全”功能。例如，我们不允许 在/home之外搜索，因为这是我们设置的根。 当您想公开文件系统时，这些功能非常有用 获得法学硕士学位。
______________________________________________________________________
快速开始
1.安装
您可以通过pip安装mcp2py：
pip install mcp2pyPython有一个棘手的问题，那就是没有一种标准的管理方式 长期依赖。为了避免依赖冲突，它是 建议使用虚拟环境。我最喜欢的方法是 随着 uv （见此处： https://docs.astral.sh/uv/getting-started/installation/).然后你就可以 创建一个新环境并安装mcp2py，如下所示：
# Install uv (if you haven't already)
curl -LsSf https://astral.sh/uv/install.sh | sh
# Create a new project with a virtual environment
uv init my-mcp-project
cd my-mcp-project
# Install mcp2py
uv add mcp2py
# Activate the environment and start coding
uv run python2.使用它
from mcp2py import load
# Load any MCP server with OAuth authentication
notion = load("https://mcp.notion.com/mcp", auth="oauth")
# Browser opens automatically for OAuth login
# Once authenticated, you can use the tools
notion.notion_get_self()3.就是这样！
服务器作为一个子进程运行，工具是Python方法，一切 只是工作。
什么是MCP？
MCP服务器暴露 工具, 资源，以及 提示 通过a 协议。mcp2py将它们转换为{python}：
- 🔧 工具 → {python}函数
- 📦 资源 → {python}常量/属性
- 📝 提示 → 模板函数/字符串
哲学
它只是工作™-但你可以自定义一切
mcp2py是为 研究人员、数据分析师和{python} 初学者 他们想尝试没有复杂性的MCP服务器。同时 时间，它提供 完全控制 为开发商建设生产 应用。
默认情况下为零配置： -OAuth登录？浏览器打开 自动-需要用户输入？出现终端提示-服务器需要 法学硕士？我们处理它——一切都是开箱即用的
高级用户没有上限： -覆盖任何默认行为- 自定义身份验证流程-构建生产应用程序-当您 需要它
您的{python}REPL/代码将成为MCP客户端。 服务器是 mcp2py通信的独立进程（Node.js、{python}等） 通过JSON-RPC。您的{python}代码可以：-调用工具（服务器 函数），就像它们是本地{python}函数一样-访问资源 （服务器数据）作为{python}属性-处理服务器请求（采样， 启发）自动或通过自定义回调-无缝工作 使用任何AI SDK（Anthropic、OpenAI、DSPy等）
入门指南
对于初学者和研究人员：它只是工作
from mcp2py import load
# Load any MCP server - that's it!
server = load("https://api.example.com/mcp")
# If it needs login:
# → Browser opens automatically
# → You log in once
# → Browser closes
# → Done!
# If it needs your input:
# → Nice terminal prompts appear
# → You answer
# → Code continues!
# If it needs AI help (sampling):
# → Uses your ANTHROPIC_API_KEY or OPENAI_API_KEY
# → Handles it automatically
# → You don't even notice!
# Just use the tools!
result = server.analyze_data(dataset="sales_2024.csv")
print(result)就是这样。没有配置。没有设置。它只是工作。
______________________________________________________________________
接口设计
基本用法
from mcp2py import load
# Load an MCP server - simple and clean
weather = load("npx -y @h1deya/mcp-server-weather")
# Or from a remote HTTP server (SSE/HTTP Stream transport)
api = load("https://api.example.com/mcp")
# With authentication
api = load("https://api.example.com/mcp", headers={"Authorization": "Bearer YOUR_TOKEN"})
# Or from a {python} script
travel = load("{python} my_mcp_server.py")
# Tools become functions
alerts = weather.get_alerts(state="CA")
forecast = weather.get_forecast(latitude=37.7749, longitude=-122.4194)
print(forecast)
# Resources become attributes
print(weather.API_DOCUMENTATION) # Constant resource
print(weather.current_config) # Dynamic resource
# Prompts become template functions
prompt = weather.create_weather_report(location="NYC", style="casual")与人工智能框架（DSPy、Claudette等）一起使用
这 .tools 属性为您提供可调用的{python}列表 函数:
from mcp2py import load
server = load("npx -y @modelcontextprotocol/server-filesystem /tmp")
# Get tools as callable functions
tools = server.tools
# [, , ...]
# Each function has __name__ and __doc__
print(tools[0].__name__) # "read_file"
print(tools[0].__doc__) # "Read a file from the filesystem"
# And they're callable!
result = tools[0](path="/tmp/test.txt")使用AI框架
这 .tools 属性为框架提供了可调用的函数 比如DSPy和克劳德特：
from mcp2py import load
import dspy
# Load MCP server
travel = load("{python} airline_server.py")
# Use with DSPy - pass callable functions directly
class CustomerService(dspy.Signature):
user_request: str = dspy.InputField()
result: str = dspy.OutputField()
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
# Pass tools directly to DSPy (it expects callables)
react = dspy.ReAct(CustomerService, tools=travel.tools)
result = react(user_request="Book a flight from SFO to JFK on 09/01/2025")
print(result)# Also works with Claudette
from mcp2py import load
from claudette import Chat
weather = load("npx -y @h1deya/mcp-server-weather")
# Claudette expects callable functions
chat = Chat(model="claude-3-5-sonnet-20241022", tools=weather.tools)
response = chat("What's the weather in Tokyo?")
# Claudette automatically calls the tools as needed
print(response)注： 对于具有原生MCP支持的SDK（Anthropic、OpenAI等）， Google Gemini），直接使用其内置的MCP集成。这 .tools 属性适用于DSPy和Claudette等框架，它们期望 {python}可调用。
类型安全和IDE支持
自动生成存根以实现完美的自动补全：
from mcp2py import load
# Stubs auto-generated to ~/.cache/mcp2py/stubs/
server = load("npx my-server")
# IDE now has full autocomplete and type hints!
server.search_files(
pattern="*.py", # type: str - IDE knows this!
max_results=10 # type: int, optional - IDE suggests this!
) # Returns: dict[str, Any] - IDE shows return type!手动生成存根：
# Generate stub to specific location for your project
server = load("npx weather-server")
server.generate_stubs("./stubs/weather.pyi")
# Or let it auto-cache (default behavior)
# Stubs saved to: ~/.cache/mcp2py/stubs/.pyi它是如何工作的： - load() 返回a 动态类型类 随着 所有方法都是预定义的-IDE会立即看到正确的类型提示- 无需配置！ -类型提示包括参数名称， 类型、默认值和返回类型-适用于VS Code、PyCharm、Jupyter 笔记本电脑和任何｛python｝IDE-也会生成 .pyi 存根文件到 ~/.cache/mcp2py/stubs/ 供参考
无需配置 -自动补全功能正常工作！ ✨
MCP客户端功能
当您的{python}代码充当MCP客户端时，服务器可能会请求这些 能力：
采样
当服务器需要LLM完成时，mcp2py会自动处理。
默认设置：开箱即用
from mcp2py import load
# Just works! Uses your default LLM
server = load("npx travel-server")
# If server needs LLM help, mcp2py:
# 1. Checks for ANTHROPIC_API_KEY or OPENAI_API_KEY in environment
# 2. Calls the LLM automatically
# 3. Returns result to server
# 4. Your code continues!
result = server.book_flight(destination="Tokyo")配置首选LLM：
# Set via environment (recommended)
import os
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
# Or configure globally using LiteLLM model strings
from mcp2py import configure
configure(
model="claude-3-5-sonnet-20241022" # or "gpt-4o", "gemini/gemini-pro", etc.
)
# LiteLLM automatically detects the right API based on model name
# Uses standard env vars: ANTHROPIC_API_KEY, OPENAI_API_KEY, etc.
# Now all servers use this LLM for sampling
server = load("npx travel-server")高级：自定义采样处理程序
from mcp2py import load
def my_sampling_handler(messages, model_prefs, system_prompt, max_tokens):
"""Full control over LLM calls."""
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 # Override default
)禁用采样（用于安全/成本控制）：
server = load(
"npx travel-server",
allow_sampling=False # Raises error if server requests LLM
)引出
当服务器需要用户输入时，mcp2py会自动提示。
默认值：终端提示
from mcp2py import load
# Just works! Terminal prompts appear automatically
server = load("npx travel-server")
# Server asks: "Confirm booking for $500?"
# Terminal shows:
#
# Server asks: Confirm booking for $500?
# confirm_booking (boolean): y/n
#
# You type: y
# Code continues!
result = server.book_flight(destination="Paris")你所看到的：
Calling book_flight...
┌─────────────────────────────────────────┐
│ 🔔 Server needs your input │
├─────────────────────────────────────────┤
│ Confirm booking for $500? │
│ │
│ confirm_booking (boolean): y/n │
│ seat_preference (window/aisle/middle): │
│ meal_preference (optional): │
└─────────────────────────────────────────┘
> y
> window
> vegetarian
Booking confirmed!高级：自定义诱导处理程序
from mcp2py import load
def my_input_handler(message, schema):
"""Custom UI for user input."""
# Build a GUI, web form, voice input, etc.
from tkinter import simpledialog
return simpledialog.askstring("Server Request", message)
server = load(
"npx travel-server",
on_elicitation=my_input_handler
)禁用启发（用于自动脚本）：
server = load(
"npx travel-server",
allow_elicitation=False # Raises error if server asks for input
)
# Or provide pre-filled answers
server = load(
"npx travel-server",
elicitation_defaults={
"confirm_booking": True,
"seat_preference": "window"
}
)根
服务器可以询问要关注哪些目录。可选，简单：
# Single directory
server = load("npx filesystem-server", roots="/home/user/projects")
# Multiple directories
server = load(
"npx filesystem-server",
roots=["/home/user/projects", "/tmp/workspace"]
)
# Update roots dynamically
server.set_roots(["/home/user/new-project"])设计规则
1. 工具→ 函数
MCP工具映射到{python}函数，完全支持：
- 参数：必需参数和可选参数
- 类型提示：从JSON模式生成
inputSchema - 文档字符串：由工具构建
description - 返回类型：键入为
dict[str, Any]（MCP工具返回JSON）
命名约定：Snake_case（MCP getWeather → python get_weather)
# MCP Tool Definition:
# {
# "name": "searchFiles",
# "description": "Search for files matching a pattern",
# "inputSchema": {
# "type": "object",
# "properties": {
# "pattern": {"type": "string", "description": "Glob pattern"},
# "maxResults": {"type": "integer", "default": 100}
# },
# "required": ["pattern"]
# }
# }
# Generated {python}:
def search_files(pattern: str, max_results: int = 100) -> dict[str, Any]:
"""Search for files matching a pattern.
Args:
pattern: Glob pattern
max_results: Maximum results to return (default: 100)
"""
...2. 资源→ 常量或属性
资源地图根据其性质而有所不同：
- 静态资源 （如文档、模式）：模块级
常数（UPPER_CASE）
- 动态资源 （可能会改变）：带有getter的属性
（小写）
# Static resource (cached)
API_DOCS: str = server._get_resource("api://docs")
# Dynamic resource (fetched on access)
@property
def current_status() -> dict[str, Any]:
"""Current server status."""
return server._get_resource("status://current")命名约定：-静态： UPPER_SNAKE_CASE -动态： lower_snake_case 属性
3. 提示→ 模板函数
提示变为返回格式化字符串的函数：
# MCP Prompt:
# {
# "name": "reviewCode",
# "description": "Generate a code review prompt",
# "arguments": [
# {"name": "code", "description": "Code to review", "required": true},
# {"name": "focus", "description": "Review focus area", "required": false}
# ]
# }
# Generated {python}:
def review_code(code: str, focus: str | None = None) -> str:
"""Generate a code review prompt.
Args:
code: Code to review
focus: Review focus area (optional)
Returns:
Formatted prompt string ready for LLM
"""
...4. 错误处理
{python}ic常见故障的例外情况：
from mcp2py.exceptions import (
MCPConnectionError, # Can't connect to server
MCPToolError, # Tool execution failed
MCPResourceError, # Resource not found
MCPValidationError, # Invalid arguments
)
try:
result = server.expensive_operation(data=large_data)
except MCPValidationError as e:
print(f"Invalid input: {e}")
except MCPToolError as e:
print(f"Tool failed: {e}")5. 异步支持
使用 aload() 对于异步MCP服务器：
from mcp2py import aload
# Async version - all tools become async
server = await aload("npx async-server")
result = await server.fetch_data(url="https://example.com")
status = await server.get_current_status()6. 上下文管理器
使用时自动清理 with:
from mcp2py import load
# Sync version
with load("npx my-server") as server:
result = server.do_work()
# Server process automatically terminated
# Async version
async with aload("npx my-server") as server:
result = await server.do_work()配置
服务器注册表（可选）
注册一次常用服务器，然后按名称加载：
from mcp2py import register, load
# Register servers (run once, e.g., in your setup script)
register(
weather="npx -y @h1deya/mcp-server-weather",
brave="npx -y brave-search-mcp-server",
filesystem="npx -y @modelcontextprotocol/server-filesystem /tmp",
myserver="{python} my_mcp_server.py"
)
# Then load by name anywhere
weather = load("weather")
brave = load("brave")
# Or use commands directly (no registration needed)
custom = load("npx my-custom-server")注册表保存到 ~/.config/mcp2py/servers.json 自动。
远程服务器和身份验证
MCP服务器可以通过HTTP远程托管（使用SSE或HTTP流 运输）：
from mcp2py import load, register
# Connect to remote MCP server
api = load("https://api.example.com/mcp")
# With Bearer token authentication
secure_api = load(
"https://api.example.com/mcp",
headers={"Authorization": "Bearer sk-1234567890"}
)
# With custom headers (API keys, etc.)
custom_api = load(
"https://api.example.com/mcp",
headers={
"X-API-Key": "your-api-key",
"X-Client-ID": "your-client-id"
}
)
# Register remote servers too
register(
production_api="https://api.prod.example.com/mcp",
staging_api="https://api.staging.example.com/mcp"
)
# Load with auth at runtime
prod = load("production_api", headers={"Authorization": f"Bearer {get_token()}"})远程MCP服务器的用例： -公司托管的内部工具- 通过MCP的付费API服务-共享团队资源（数据库、分析、， 等）-基于云的人工智能工具市场
OAuth身份验证（谷歌、GitHub等）
默认值：零配置（适用于初学者、研究人员、数据 分析师）
mcp2py自动处理OAuth-只需加载并运行：
from mcp2py import load
# That's it! Browser opens, you log in, then continue coding
server = load("https://api.example.com/mcp")
# First tool call triggers OAuth if needed:
# 1. Browser window pops up
# 2. You log in (Google/GitHub/etc.)
# 3. Window closes automatically
# 4. Your code continues!
result = server.my_tool() # Works immediately after login引擎盖下发生了什么： -mcp2py检测OAuth需求（401 response）-自动发现OAuth端点-打开浏览器 登录（PKCE安全）-将令牌存储在 ~/.config/mcp2py/tokens.json - 令牌过期时自动刷新
你从不考虑代币。
______________________________________________________________________
高级：自定义OAuth（用于生产应用程序）
构建应用程序时覆盖默认值：
from mcp2py import load
# Option 1: Custom token provider
def get_google_token():
"""Your custom OAuth logic."""
from google.oauth2.credentials import Credentials
# Your implementation here
return creds.token
server = load(
"https://api.example.com/mcp",
auth=get_google_token # Called when token needed
)
# Option 2: Service account (no browser)
from google.oauth2 import service_account
credentials = service_account.Credentials.from_service_account_file(
'service-account.json'
)
server = load(
"https://api.example.com/mcp",
auth=credentials
)
# Option 3: Manual token management
server = load(
"https://api.example.com/mcp",
headers={"Authorization": f"Bearer {your_token}"}
)
# Option 4: Disable auto-browser (for servers/CI)
server = load(
"https://api.example.com/mcp",
auto_auth=False # Raises error instead of opening browser
)环境变量支持（用于生产）：
# Set token via environment
export MCP_TOKEN="your-token-here"# Automatically used if available
server = load("https://api.example.com/mcp")安全注意事项
客户端（mcp2py自动处理）： - ✅ 安全令牌 存储-OAuth令牌缓存在 ~/.fastmcp/oauth-mcp-client-cache/ - ✅ OAuth流的PKCE支持（代码交换的证明密钥）-✅ 到期前自动刷新令牌-✅ 环境变量 支持(MCP_TOKEN)
服务器端（连接时由您负责）： -使用HTTPS URL 对于生产服务器（非HTTP）-确保您连接的MCP服务器 要实现正确的身份验证，请旋转令牌/凭据 定期-从不将令牌提交到版本控制
最佳实践：
# Good: Use environment variables
import os
server = load("https://api.example.com/mcp", auth=os.getenv("MCP_TOKEN"))
# Good: HTTPS for production
server = load("https://api.example.com/mcp", auth="oauth")
# Avoid: Hardcoded tokens in code
# server = load("https://api.example.com/mcp", auth="sk-secret-123") # Don't do this!高级功能
树桩生成
使用时会自动生成存根 load()它们已被缓存 到 ~/.cache/mcp2py/stubs/ 以供重复使用。
编程API：
from mcp2py import load
# Stubs auto-generated on load
server = load("npx weather-server")
# Generate to specific path
stub_path = server.generate_stubs("./stubs/weather.pyi")
print(f"Stub saved to: {stub_path}")
# Check cache location
from mcp2py.stubs import get_stub_cache_path
cache_path = get_stub_cache_path("npx weather-server")
print(f"Cached at: {cache_path}")完整客户端示例
"""Full example of {python} as MCP client with all features."""
from mcp2py import load
import anthropic
# Setup callbacks for server requests
def handle_sampling(messages, model_prefs, system_prompt, max_tokens):
"""Server wants LLM completion."""
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
messages=messages,
system=system_prompt,
max_tokens=max_tokens
)
return response.content[0].text
def handle_elicitation(message, schema):
"""Server needs user input."""
print(f"\n🔔 Server asks: {message}")
if schema.get("type") == "string":
return input("→ ")
if schema.get("type") == "boolean":
return input("→ (y/n): ").lower() in ["y", "yes", "true"]
if schema.get("type") == "object":
result = {}
for prop, details in schema.get("properties", {}).items():
result[prop] = input(f" {prop} ({details.get('description', '')}): ")
return result
import json
return json.loads(input("→ (JSON): "))
# Connect to server with all features
server = load(
"npx travel-booking-server",
on_sampling=handle_sampling,
on_elicitation=handle_elicitation,
roots="/home/user/travel-docs"
)
# Use the server - callbacks invoked automatically when needed
booking = server.book_flight(destination="Barcelona", dates="June 15-22")
print(booking)检查
from mcp2py import load
server = load("npx my-server")
# List all available tools
print(server.tools) # List of tool schemas for AI SDKs
# Get tool info
print(server.get_weather.__doc__)
print(server.get_weather.__signature__)
# List resources
print(server.resources)
# List prompts
print(server.prompts)中间件和钩子
from mcp2py import load
def log_tool_calls(tool_name: str, args: dict, result: dict):
print(f"Called {tool_name} with {args} → {result}")
server = load(
"npx my-server",
on_tool_call=log_tool_calls,
timeout=30.0
)实施优先级
第一阶段：核心功能
load()使用stdio传输功能- Tool → 带类型提示的函数映射
- 简单的资源访问
- 提示→ 模板函数映射
.toolsAI SDK集成的属性
第二阶段：开发者体验
- 生成Stub以支持IDE
- 服务器注册表(
~/.config/mcp2py/servers.json) - 上下文管理器协议
- 更好的错误消息和异常
第三阶段：高级功能
aload()用于异步支持- HTTP服务器的SSE传输
- 中间件/挂钩系统
- 采样和启发回调
设计原则
- 令人愉快的违约：身份验证、采样、启发所有
自动工作
- 没有上限：每个默认值都可以被覆盖以供生产使用
案例
- 初学者友好：数据分析师和研究人员可以开始
立即
- 生产就绪：开发人员完全控制构建应用程序
- 渐进式披露：默认情况下简单，当你
需要它
- 类型安全：尽可能生成类型以支持IDE
- {python}ic：将MCP约定转换为{python}约定
自动地
- 清除错误：当事情出错时，提供有用的信息
建议
完整示例
示例1：使用DSPy进行同步天气分析
#!/usr/bin/env {python}3
"""Analyze weather alerts using DSPy and MCP."""
from mcp2py import load
import dspy
# Configure DSPy
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
# Load MCP weather server
weather = load("npx -y @h1deya/mcp-server-weather")
# Define DSPy signature
class WeatherAnalyzer(dspy.Signature):
"""Analyze weather alerts and provide recommendations."""
state: str = dspy.InputField()
analysis: str = dspy.OutputField(desc="Weather analysis and travel recommendations")
# Create agent with MCP tools
agent = dspy.ReAct(WeatherAnalyzer, tools=weather.tools)
# Analyze weather for multiple states
states = ["CA", "NY", "TX", "FL"]
for state in states:
# Agent automatically calls weather.get_alerts() and weather.get_forecast()
result = agent(state=state)
print(f"\n{state}:")
print(result.analysis)示例2：异步旅行预订系统
#!/usr/bin/env {python}3
"""Async travel booking system with MCP and Anthropic."""
import asyncio
from mcp2py import aload
import anthropic
async def book_trip(user_request: str):
"""Book a trip using MCP travel server and Claude."""
# Load async MCP server
travel = await aload("{python} travel_server.py")
# Setup Anthropic client
client = anthropic.Anthropic()
# Initial request to Claude with MCP tools
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=2048,
tools=travel.tools, # MCP tools passed to Claude
messages=[{"role": "user", "content": user_request}]
)
# Handle tool calls in a loop
messages = [{"role": "user", "content": user_request}]
while response.stop_reason == "tool_use":
# Extract tool calls from response
tool_results = []
for content_block in response.content:
if content_block.type == "tool_use":
# Call MCP tool asynchronously
tool_name = content_block.name
tool_args = content_block.input
print(f"Calling {tool_name}({tool_args})...")
# Execute tool via MCP
tool_func = getattr(travel, tool_name)
result = await tool_func(**tool_args)
tool_results.append({
"type": "tool_result",
"tool_use_id": content_block.id,
"content": str(result)
})
# Add assistant response and tool results to conversation
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
# Continue conversation
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=2048,
tools=travel.tools,
messages=messages
)
# Extract final response
return response.content[0].text
async def main():
result = await book_trip(
"Book a round-trip flight from SFO to JFK on Sept 1-8, 2025. "
"My name is Adam Smith. I prefer window seats and morning flights."
)
print("\n" + "="*60)
print("BOOKING RESULT:")
print("="*60)
print(result)
if __name__ == "__main__":
asyncio.run(main())示例3：简单的同步-直接工具调用
#!/usr/bin/env {python}3
"""Simple weather check without AI - just direct MCP tool calls."""
from mcp2py import load
# Load weather server
weather = load("npx -y @h1deya/mcp-server-weather")
# Direct tool calls (no LLM needed)
print("Weather Alerts for California:")
alerts = weather.get_alerts(state="CA")
print(alerts)
print("\nSan Francisco Forecast:")
forecast = weather.get_forecast(latitude=37.7749, longitude=-122.4194)
print(forecast)
# MCP tools are just {python} functions!使用真实服务器进行测试
这里是 您现在可以测试的真实MCP服务器:
from mcp2py import load
# Weather server (Node.js via npx)
weather = load("npx -y @h1deya/mcp-server-weather")
# Brave search (requires API key)
brave = load("npx -y brave-search-mcp-server")
# Filesystem operations
fs = load("npx -y @modelcontextprotocol/server-filesystem /tmp")
# Memory/knowledge graph
memory = load("npx -y @modelcontextprotocol/server-memory")
# Remote HTTP server
api = load("https://api.example.com/mcp")
# Remote server with authentication
secure_api = load(
"https://api.example.com/mcp",
headers={"Authorization": "Bearer YOUR_TOKEN"}
)
# Inspect what's available
print(weather.tools) # List of tool schemas
print(weather.get_alerts) # Callable function
result = weather.get_alerts(state="CA")干净、简单、，{python}ic.这就是目标。 🎯
______________________________________________________________________
架构概述
┌─────────────────────────────────────────────────────────────────┐
│ Your {python} Code (MCP Client) │
│ │
│ from mcp2py import load │
│ │
│ server = load("npx weather-server") │
│ result = server.get_forecast(lat=37.7, lon=-122.4) │
│ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ Optional: Use with AI SDKs │ │
│ │ │ │
│ │ import dspy │ │
│ │ agent = dspy.ReAct( │ │
│ │ Signature, │ │
│ │ tools=server.tools # ← mcp2py │ │
│ │ ) │ │
│ └──────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
↕ JSON-RPC over stdio
┌─────────────────────────────────────────────────────────────────┐
│ MCP Server Process (separate process) │
│ │
│ Node.js / {python} / Rust / whatever │
│ Exposes: tools, resources, prompts │
│ May request: sampling, elicitation, roots │
└─────────────────────────────────────────────────────────────────┘要点： 1. mcp2py是客户端 -它与JSON-RPC通信 服务器2。 服务器是一个独立的进程 -开始于 command 参数3。 低级和通用 -适用于任何AI SDK或 独立4。 双向 -客户端调用服务器工具，服务器可以 请求客户端功能
______________________________________________________________________
高级教程：使用DSPy构建MCP服务器
本教程通过展示如何演示mcp2py的全部功能 至：1。在DSPy代理2中使用Notion MCP。将该代理包装为自己的MCP 服务器3。在另一个DSPy程序中使用该服务器
这证明了事实 可组合性 -通过以下方式构建复杂的人工智能系统 将MCP服务器链接在一起。
步骤1:带有Notion MCP的DSPy代理
首先，让我们创建一个可以搜索和组织的DSPy代理 Notion中的信息：
import dspy
from mcp2py import load
# Configure DSPy with your LLM
dspy.configure(lm=dspy.LM("openai/gpt-4.1"))
# Load Notion MCP server
notion = load("https://mcp.notion.com/mcp", auth="oauth")
# Define a DSPy signature for a research assistant
class NotionResearcher(dspy.Signature):
"""Research assistant that searches Notion workspace."""
query: str = dspy.InputField(desc="Research query")
summary: str = dspy.OutputField(desc="Summary of findings")
# Create DSPy agent with Notion tools
researcher = dspy.ReAct(NotionResearcher, tools=notion.tools)
# Test the agent
result = researcher(query="What are our Q1 2025 goals?")
print(result.summary)[11/03/25 07:20:44] INFO OAuth authorization URL: oauth.py:340
https://mcp.notion.com/authorizresponse_typ...
INFO 🎧 OAuth callback server started on http://localhost:40621 oauth.py:358
AI response > No information was found regarding our Q1 2025 goals due to repeated execution errors when attempting to search or create documentation in Notion. All attempts to gather relevant data or create a new page were unsuccessful.步骤2：将代理包装为MCP服务器
现在，让我们使用以下命令将这个由Notion驱动的代理转换为自己的MCP服务器 FastMCP：
# notion_research_server.py
from fastmcp import FastMCP
from mcp2py import load
import dspy
# Initialize FastMCP
mcp = FastMCP("Notion Research Server")
# Load Notion and configure DSPy
notion = load("https://mcp.notion.com/mcp", auth="oauth")
dspy.configure(lm=dspy.LM("openai/gpt-4.1"))
class NotionResearcher(dspy.Signature):
"""Research assistant that searches Notion workspace."""
query: str = dspy.InputField(desc="Research query")
summary: str = dspy.OutputField(desc="Summary of findings")
researcher = dspy.ReAct(NotionResearcher, tools=notion.tools)
@mcp.tool()
def research_notion(query: str) -> str:
"""Research a topic by searching the Notion workspace.
Args:
query: The research question or topic to investigate
Returns:
A comprehensive summary of findings from Notion
"""
result = researcher(query=query)
return result.summary
# Run the server
if __name__ == "__main__":
mcp.run()将此另存为 notion_research_server.py 现在您有了一个自定义MCP 服务器！
步骤3：在另一个DSPy程序中使用自定义服务器
最后，让我们在更高级别上使用我们的自定义Notion Research Server DSPy代理：
import dspy
from mcp2py import load
# Configure DSPy
dspy.configure(lm=dspy.LM("openai/gpt-4.1"))
# Load our custom Notion Research Server
research_server = load("uv run notion_research_server.py")
# Create a high-level report writer
class ReportWriter(dspy.Signature):
"""Executive assistant that creates comprehensive reports."""
topic: str = dspy.InputField(desc="Report topic")
sections: str = dspy.InputField(desc="Comma-separated section topics")
report: str = dspy.OutputField(desc="Comprehensive markdown report")
# Create agent with our research server tools
report_writer = dspy.ReAct(ReportWriter, tools=research_server.tools)
# Generate a comprehensive report
result = report_writer(
topic="Company Strategy Review",
sections="Q1 Goals, Recent Achievements, Upcoming Initiatives, Team Updates"
)
print("="*60)
print(result.report)
print("="*60)============================================================
# Company Strategy Review
## Q1 Goals
*Unable to retrieve specific information due to access issues.*
## Recent Achievements
*Unable to retrieve specific information due to access issues.*
## Upcoming Initiatives
*Unable to retrieve specific information due to access issues.*
## Team Updates
*Unable to retrieve specific information due to access issues.*
---
*Note: This report could not be completed with current data access. Please provide the necessary information or resolve access issues to enable a comprehensive review.*
============================================================刚刚发生了什么？
此示例演示 三个层次的构成:
- 级别1：Notion MCP提供原始工具（搜索、获取、创建）
- 2级：DSPy代理+Notion MCP=研究助理（包裹
作为MCP服务器）
- 级别3：另一个DSPy代理使用Research Assistant
创建综合报告
关键利益
- 模块化：每一层都是独立的，可重复使用
- 抽象：更高级别不需要了解Notion的API
- 可组合性：混合搭配不同的MCP服务器和代理
- 类型安全：整个链中的完整类型提示和IDE支持
- 可测试性：每一层都可以独立测试
实际应用
此模式支持强大的工作流：
- 多源研究：结合Notion、GitHub、Slack和其他
MCP服务器
- 专业代理：创建特定领域的助理（HR，
工程、销售）
- 代理链：构建复杂的管道，让代理人呼叫其他人
代理
- API摘要：将复杂性隐藏在简单、高级的工具后面
后续步骤
尝试创建自己的组合系统：
- 添加更多MCP服务器（GitHub、Slack、Google Drive）
- 为不同领域创建专门的代理
- 与主管和员工建立代理层次结构
- 作为微服务部署以供生产使用
当您可以将任何MCP服务器转换为 将Python和任何Python代码导入MCP服务器！