MCP服务器SDK

用于简化MCP服务器(模型上下文协议)创建的Python SDK,支持通过装饰器和适配器快速构建和集成工具。

  • by tlcharchar
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 tlcharchar/mcp-toolkit

使用说明

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

贡献

  1. 克隆该仓库
  2. 创建您的分支(git checkout -b feature/new-feature)
  3. 提交您的更改(git commit -m 'Add new feature')
  4. 推到分支(git push origin feature/new-feature)
  5. 打开拉取请求

许可证

MIT许可证——见 许可证 了解详情。

链接