使用说明
MCP工具包
用于简化MCP服务器创建的Python SDK(模型上下文协议)。
概述
MCP Toolkit允许开发人员以声明和Python方式构建MCP服务器,并提供一流的支持,将现有的REST API作为MCP工具公开。
主要特点
- 装饰师优先:Pythonic API使用
@tool,@resource,@prompt - REST优先:内置适配器,用于使用REST API并将其作为工具公开
- OpenAPI导入:根据OpenAPI规范自动生成工具
- 类型安全:广泛使用Pydantic进行验证和模式生成
- 可观察对象:结构化日志记录和可选的Prometheus指标
安装
pip install mcp-toolkit
对于可选功能:
# OpenAPI support
pip install mcp-toolkit[openapi]
# Prometheus metrics
pip install mcp-toolkit[metrics]
# Everything included
pip install mcp-toolkit[all]
快速开始
示例1:基本服务器
from mcp_toolkit import MCPServer
server = MCPServer("my_mcp")
@server.tool()
def calculate_shipping(origin_zip: str, destination_zip: str, weight_kg: float) -> dict:
"""Calculates shipping cost between two zip codes."""
return {"cost": 45.90, "delivery_days": 3}
if __name__ == "__main__":
server.run()
示例2:公开REST API
from mcp_toolkit import MCPServer
from mcp_toolkit.adapters import RestApiAdapter
from mcp_toolkit.auth import bearer
server = MCPServer("sales_mcp")
api = RestApiAdapter(
base_url="https://api.company.com/v1",
auth=bearer("API_TOKEN"), # Reads from env var API_TOKEN
)
# Map endpoint → tool
server.register_tool(
api.as_tool(
method="GET",
path="/customers/{customer_id}",
name="get_customer",
description="Fetches customer data by ID",
)
)
# Generate full CRUD
server.register_tools(
api.crud_tools(
resource_name="order",
resource_path="/orders",
)
)
server.run()
示例3:从OpenAPI导入
from mcp_toolkit import MCPServer
from mcp_toolkit.adapters import OpenAPIAdapter
server = MCPServer("crm_mcp")
adapter = OpenAPIAdapter.from_url(
"https://api.crm.company.com/openapi.json",
auth_header="X-API-Key",
# optional base_url: use if the spec has no "servers" or to override it
# base_url="http://localhost:8000",
)
# Register all operations
server.register_tools(adapter.all_tools())
# Or filter by tags
server.register_tools(
adapter.tools_matching(tags=["customers", "opportunities"])
)
server.run()
文档
MCP服务器
创建MCP服务器的主类。
from mcp_toolkit import MCPServer, ServerConfig, TransportType
# Simple form
server = MCPServer("my_mcp")
# With detailed configuration
config = ServerConfig(
name="my_mcp",
version="1.0.0",
transport=TransportType.STREAMABLE_HTTP,
port=9000,
log_level="DEBUG",
)
server = MCPServer(config)
传输(stdio、SSE、流式HTTP)
SDK通过以下方式支持三种传输类型 TransportType 和 server.run():
| 运输 | 使用 | 何时使用 |
|---|---|---|
| 工作室 (默认) | server.run() 或 transport=TransportType.STDIO |
客户端(例如Cursor)启动进程并通过stdin/stdout进行通信。非常适合当地使用。 |
| 上海证券交易所 | server.run(transport=TransportType.SSE, host="127.0.0.1", port=9000) |
带有服务器发送事件的HTTP服务器。终点: /sse, /messages/. |
| 流式HTTP | server.run(transport=TransportType.STREAMABLE_HTTP, port=9000) |
现代HTTP服务器,单端点(例如。 /mcp).建议用于网络/云部署。 |
对于HTTP传输(SSE或Streamable), host 和 port 来自 ServerConfig 或者可以通过 run(host=..., port=...).
装饰器
@工具
将函数注册为MCP工具。
from pydantic import BaseModel, Field
# Simple form
@server.tool()
def my_tool(param: str) -> str:
"""Tool description."""
return f"Result: {param}"
# With Pydantic model
class InputModel(BaseModel):
name: str = Field(..., description="User name")
age: int = Field(..., ge=0, le=150, description="Age")
@server.tool()
def create_user(params: InputModel) -> dict:
"""Creates a new user."""
return {"name": params.name, "age": params.age}
# With annotations
@server.tool(
name="fetch_data",
annotations={"readOnlyHint": True, "idempotentHint": True}
)
def fetch(id: str) -> dict:
"""Fetches data by ID."""
return {"id": id}
@资源
注册可通过URI访问的资源。
@server.resource(uri="docs://{name}")
def get_document(name: str) -> str:
"""Returns document content."""
return Path(f"docs/{name}.md").read_text()
@server.resource(uri="config://{key}", mime_type="application/json")
def get_config(key: str) -> str:
"""Returns configuration."""
return json.dumps(configs.get(key))
@提示
注册提示模板。
@server.prompt(name="customer-analysis")
def analysis_prompt(customer_id: str) -> str:
"""Prompt for customer analysis."""
return f"""
Analyze customer {customer_id} considering:
1. Purchase history
2. Support tickets
3. Upsell potential
"""
REST适配器
将REST端点转换为MCP工具。
from mcp_toolkit.adapters import RestApiAdapter
from mcp_toolkit import ToolParameter
api = RestApiAdapter(
base_url="https://api.example.com",
auth=bearer("TOKEN"),
timeout=30.0,
max_retries=3,
)
# Specific endpoint
tool = api.as_tool(
method="GET",
path="/users/{id}",
name="get_user",
description="Fetches user by ID",
query_params=[
ToolParameter(
name="fields",
type="string",
description="Fields to return",
required=False,
)
],
)
# Full CRUD
tools = api.crud_tools(
resource_name="product",
resource_path="/products",
create_params=[
ToolParameter(name="name", type="string", description="Name", required=True),
ToolParameter(name="price", type="number", description="Price", required=True),
],
)
OpenAPI适配器
自动导入OpenAPI规范。
- 基础URL:规格应该有
servers: [{"url": "https://..."}]如果缺少完整的URL,适配器将从规范URL派生它(例如。http://localhost:8000/openapi.json→http://localhost:8000)或者你可以通过base_url在from_url().
from mcp_toolkit.adapters import OpenAPIAdapter
# From URL (base_url derived from spec URL if spec has no "servers")
adapter = OpenAPIAdapter.from_url("https://api.com/openapi.json")
# With explicit base_url (useful when spec doesn't define servers)
adapter = OpenAPIAdapter.from_url(
"http://localhost:8000/openapi.json",
base_url="http://localhost:8000",
)
# From file (base_url required)
adapter = OpenAPIAdapter.from_file("./openapi.yaml", base_url="https://api.com")
# List available operations
print(adapter.list_available_tags())
print(adapter.list_available_operations())
# Filter tools
tools = adapter.tools_matching(
tags=["users"],
methods=["GET", "POST"],
path_pattern=r"/api/v2/.*"
)
认证
用于配置身份验证的帮助程序。
from mcp_toolkit.auth import api_key, bearer, basic, oauth2
# API Key
auth = api_key("X-API-Key", env_var="MY_API_KEY")
# Bearer token
auth = bearer("MY_TOKEN")
# Basic auth
auth = basic("USERNAME_VAR", "PASSWORD_VAR")
# OAuth2
auth = oauth2(
client_id_env="CLIENT_ID",
client_secret_env="CLIENT_SECRET",
token_url="https://auth.com/oauth/token",
scopes=["read", "write"],
)
可观测性
结构化日志记录和指标。
from mcp_toolkit.observability import ToolkitLogger, MetricsCollector
# Logging
logger = ToolkitLogger("my_mcp", level="DEBUG", structured=True)
logger.info("Message", extra_field="value")
# Metrics (requires prometheus-client)
metrics = MetricsCollector(prefix="my_mcp")
metrics.start_server(port=9090)
with metrics.measure_tool("my_tool"):
# tool execution
pass
命令行界面
该工具包包括一个用于脚手架的CLI。
# Create new project
mcp-toolkit init my_server
# With REST template
mcp-toolkit init my_server --template rest --base-url https://api.com
# REST without authentication (local APIs)
mcp-toolkit init my_server --template rest --base-url http://localhost:8000 --no-auth
# With OpenAPI template (--base-url optional; --no-auth for unauthenticated APIs)
mcp-toolkit init my_server --template openapi --openapi-url http://localhost:8000/openapi.json --base-url http://localhost:8000
mcp-toolkit init my_server_mcp --template openapi --openapi-url http://localhost:8000/openapi.json --base-url http://localhost:8000 --no-auth
# Validate server
mcp-toolkit validate server.py
# List tools
mcp-toolkit list-tools server.py --format json
推荐项目结构
my_mcp/
├── server.py # Main server
├── tools/ # Custom tools
│ ├── __init__.py
│ ├── sales.py
│ └── inventory.py
├── resources/ # Resources
│ └── docs/
├── config.py # Configuration
├── requirements.txt
└── README.md
测试
# Install dev dependencies
pip install mcp-toolkit[dev]
# Run tests
pytest
# With coverage
pytest --cov=mcp_toolkit
贡献
- 克隆该仓库
- 创建您的分支(
git checkout -b feature/new-feature) - 提交您的更改(
git commit -m 'Add new feature') - 推到分支(
git push origin feature/new-feature) - 打开拉取请求
许可证
MIT许可证——见 许可证 了解详情。
