MCP Runtime (Hassansaleem Aptly)

MCP运行时

模型上下文协议的生产级Python运行时

   

*由以下人员建造和维护 哈桑·萨利姆*

______________________________________________________________________

什么是MCPRuntime？

MCP运行时 是一个用于构建、部署和管理的生产级Python SDK 模型上下文协议（MCP） 服务器——开放标准 它将人工智能代理（Claude、GPT-4等）与现实世界的工具和数据连接起来。

你编写Python函数。MCPRuntime在几秒钟内将它们暴露给任何兼容MCP的AI客户端， 已经内置了生产级功能：生命周期挂钩、速率限制、实时指标， 健康端点和模块化注册表系统。

这是给谁的？

• 人工智能工程师 构建工具增强LLM代理
• 后端开发人员 将Python服务与n8n、Claude Desktop或自定义AI客户端集成
• 平台团队 微服务堆栈需要一个可观察的、生产就绪的MCP运行时

______________________________________________________________________

特性

______________________________________________________________________

安装

pip install mcpruntime
# or with uv (recommended)
uv add mcpruntime

# With CLI tools
pip install "mcpruntime[cli]"

# With WebSocket support
pip install "mcpruntime[cli,ws]"

______________________________________________________________________

快速开始

# server.py
from mcpruntime.server.mcpserver import MCPServer

server = MCPServer("my-agent", instructions="A helpful automation agent")

@server.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

@server.tool()
def greet(name: str, formal: bool = False) -> str:
    """Generate a greeting."""
    return f"Good day, {name}." if formal else f"Hey {name}!"

@server.resource("mcpruntime://config")
def get_config() -> str:
    """Return current server configuration."""
    return '{"version": "1.0", "env": "production"}'

if __name__ == "__main__":
    server.run()

mcpruntime dev server.py       # Launch with MCP Inspector
mcpruntime run server.py       # Run via stdio
mcpruntime list-tools server.py

______________________________________________________________________

生产特点

生命周期挂钩

在每次工具调用之前/之后运行代码——不更改工具函数：

from mcpruntime.server.mcpserver import MCPServer, HookManager

server = MCPServer("my-agent")
hooks = HookManager(server)

@hooks.before_tool()
async def audit(tool_name: str, arguments: dict) -> None:
    print(f"→ {tool_name}({arguments})")

@hooks.after_tool()
async def log_result(tool_name: str, result, duration_ms: float) -> None:
    print(f"← {tool_name} returned in {duration_ms:.1f}ms")

@hooks.on_error()
async def alert(tool_name: str, error: Exception) -> None:
    print(f"✗ {tool_name} raised: {error}")

速率限制

通过每个客户、每个工具的费率限制来保护您的工具免受滥用：

from mcpruntime.server.mcpserver import MCPServer, RateLimitMiddleware

server = MCPServer("my-agent")
limiter = RateLimitMiddleware(max_calls=10, window_seconds=60)
limiter.install(server)

实时指标

跟踪呼叫计数和错误率；将它们作为MCP资源公开：

from mcpruntime.server.mcpserver import MCPServer, MetricsMiddleware

server = MCPServer("my-agent")
metrics = MetricsMiddleware()
metrics.install(server)

@server.resource("mcpruntime://metrics")
def show_metrics() -> str:
    return metrics.get_report()

报告看起来像：

MCPRuntime Metrics Report
Uptime : 3842s

Tool                             Calls   Errors
──────────────────────────────────────────────
  add                              142        0
  greet                             87        2
  query_db                          31        1

健康终点

添加 /health, /ready，以及 /info 只需一次调用即可连接到任何HTTP服务器：

from mcpruntime.server.mcpserver import MCPServer
from mcpruntime.server.mcpserver.health import install_health_routes

server = MCPServer("my-agent")
install_health_routes(server)
server.run(transport="streamable-http", port=8000)

GET /health  →  {"status": "ok", "uptime_seconds": 42.1, "sdk": "mcpruntime"}
GET /ready   →  {"status": "ready"}
GET /info    →  {"sdk": "mcpruntime", "version": "1.0.0", "server_name": "my-agent"}

模块注册表

跨文件拆分工具，在启动时组合它们：

# tools/search.py
from mcpruntime.server.mcpserver.registry import registry

@registry.tool()
def web_search(query: str) -> str:
    """Search the web."""
    ...

# tools/database.py
from mcpruntime.server.mcpserver.registry import registry

@registry.tool()
def query_db(sql: str) -> str:
    """Run a read-only SQL query."""
    ...

# server.py
from mcpruntime.server.mcpserver import MCPServer
from mcpruntime.server.mcpserver.registry import registry
import tools.search, tools.database

server = MCPServer("my-agent")
registry.mount(server)   # registers all tools from both modules
server.run()

结构化输出

返回Pydantic模型——客户端收到经过验证的、模式强制的JSON：

from pydantic import BaseModel

class WeatherResult(BaseModel):
    city: str
    temperature_c: float
    condition: str

@server.tool()
def get_weather(city: str) -> WeatherResult:
    """Get current weather for a city."""
    return WeatherResult(city=city, temperature_c=22.5, condition="sunny")

上下文——日志记录和进度

from mcpruntime.server.mcpserver import MCPServer, Context

server = MCPServer("my-agent")

@server.tool()
async def process_data(path: str, ctx: Context) -> str:
    await ctx.info(f"Processing: {path}")
    await ctx.report_progress(0, 100)
    # ... work ...
    await ctx.report_progress(100, 100)
    return "done"

______________________________________________________________________

运输

# stdio (default — Claude Desktop, MCP clients)
server.run()

# Streamable HTTP (n8n, web clients, microservices)
server.run(transport="streamable-http", host="0.0.0.0", port=8000)

# SSE (legacy HTTP transport)
server.run(transport="sse", port=8000)

______________________________________________________________________

CLI 参考

MCPRuntime附带了两个CLI别名： mcpruntime （完整）和 mr （简称）。

# Run on HTTP with custom port
mcpruntime run server.py --transport streamable-http --port 9000

# Dev mode with extra packages
mcpruntime dev server.py --with httpx --with pandas

# Set env vars inline
mcpruntime run server.py --env API_KEY=abc --env DEBUG=1

# Load a .env file
mcpruntime run server.py --env-file .env.production

# Specify which object to use
mcpruntime run server.py:my_server

______________________________________________________________________

n8n集成

1. 安装MCPRuntime： pip install "mcpruntime[cli]"
2. 写你的服务器： server.py
3. 运行它： mcpruntime run server.py --transport streamable-http --port 8000
4. 在n8n中，添加一个 MCP客户端 指向的节点 http://your-server:8000/mcp

您的Python工具现在可以从任何n8n工作流调用。

______________________________________________________________________

真实世界用例

• AI代理工具 --允许Claude或GPT-4访问您的内部API、数据库或文件系统
• n8n自动化 --将Python逻辑作为n8n工作流中的HTTP可调用工具公开
• 内部开发平台 --构建一个共享的工具注册表，你的整个团队的人工智能代理都可以调用
• 微服务Sidecar --将MCP兼容的工具层添加到任何现有的Python服务中

______________________________________________________________________

需求

• Python 3.10+
• 适用于Linux、macOS、Windows

______________________________________________________________________

项目结构

src/mcpruntime/
├── server/
│   ├── mcpserver/
│   │   ├── server.py        # MCPServer — main high-level API
│   │   ├── hooks.py         # Lifecycle hook manager
│   │   ├── middleware.py    # Rate limiting, timing, metrics
│   │   ├── health.py        # Health check HTTP routes
│   │   ├── registry.py      # Modular tool registry
│   │   ├── tools/           # Tool registration + management
│   │   ├── resources/       # Resource registration + templates
│   │   ├── prompts/         # Prompt registration
│   │   ├── auth/            # OAuth 2.0 authentication
│   │   └── utilities/       # Logging, type helpers
│   ├── lowlevel/            # Low-level MCP protocol server
│   └── auth/                # Bearer token + OAuth middleware
├── client/                  # MCP client implementation
├── shared/                  # Protocol types, session, transport
├── types/                   # MCP type definitions (JSON-RPC, etc.)
├── cli/                     # typer-based CLI
└── os/                      # Platform utilities (POSIX / Win32)

______________________________________________________________________

作者

哈桑·萨利姆 github： @哈桑萨利姆恰如其分 存储库： 哈桑萨雷姆恰如其分地/麦克普伦蒂姆

______________________________________________________________________

许可证

麻省理工学院许可证--版权所有（c）2026 Hassan Saleem