使用说明
MCP服务器框架
用于构建MCP(模型上下文协议)服务器的Python框架。 实现MCP 1.26核心协议(工具、资源、提示、通知)。 可选功能(完成、订阅、进度、取消)尚未通过插件API公开。
PyPI:
pip install mcp-server-framework--回购名称为mcp_server_toolkit由于历史原因。
三个包,一个生态系统:
| 包 | 角色 |
|---|---|
| mcp_server_框架 | 共享库:配置、传输、健康、插件基础设施 |
| mcp服务器工厂 | 启动时加载静态插件(CLI工具) |
| mcpserver_proxy | 运行时动态加载插件(具有管理API的守护进程) |
安装
pip install mcp-server-framework
发展:
pip install -e ".[dev]"
Factory——从插件构建静态服务器
mcp-factory --plugins echo --http 12201 --plugin-dir ./plugins
代理——具有运行时插件管理的动态服务器
# Start proxy with echo plugin on HTTP
mcp-proxy serve --autoload echo --http 12200 --plugin-dir ./plugins
# Manage plugins at runtime via CLI
mcp-proxy status
mcp-proxy load greet
mcp-proxy unload echo
mcp-proxy reload greet
插件也可以通过MCP工具进行管理(proxy__load, proxy__unload, proxy__reload)
或REST管理API-来自任何连接的客户端。
框架——构建一个独立的服务器
from mcp_server_framework import load_config, create_server, run_server
config = load_config()
mcp = create_server(config)
@mcp.tool()
def hello(name: str) -> str:
"""Says hello."""
return f"Hello, {name}!"
run_server(mcp, config)
插件接口
每个插件都实现一个功能:
def register(mcp, config: dict) -> None:
@mcp.tool()
def my_tool(param: str) -> str:
"""Tool description for the LLM."""
return do_something(param)
@mcp.resource("myns://status")
def my_resource() -> str:
return "OK"
@mcp.prompt()
def my_prompt(text: str) -> str:
return f"Analyze: {text}"
工具、资源和提示都会被跟踪并检查是否存在冲突。
适用于Factory和Proxy,无需更改。
看 plugins/demo_full.py 举一个完整的例子。
插件加载
该框架支持两个插件源:
本地插件 --中的文件或目录 --plugin-dir:
mcp-proxy serve --autoload echo --plugin-dir ./plugins
已安装的软件包 --点状Python导入:
autoload:
- echo # local plugin from --plugin-dir
- mcp_shell_tools.shell # installed PyPI package
- mcp_wekan_tools.wekan # installed PyPI package
点名称直接通过以下方式导入 importlib。简单的名称会通过插件目录搜索。
创建新插件
./scripts/new-plugin.sh myservice # creates plugins/myservice/
./scripts/new-plugin.sh myservice ./src # custom target directory
生成 __init__.py (MCP接线)和 tools.py (纯逻辑)带注释
示例--开始编辑 tools.py,重新启动代理,完成。
推荐插件结构
plugins/myservice/
├── __init__.py # register(mcp, config) — thin MCP wrapper
├── client.py # HTTP client (pure Python, no MCP)
└── tools.py # Tool logic (pure Python, no MCP)
这种分离使您的业务逻辑可测试和可重用 没有MCP依赖性。
演示插件
| 插件 | 工具 | 说明 |
|---|---|---|
echo |
2 | 最小示例——echo和echo_upper |
greet |
1 | 最小示例——按名字问候 |
demo_full |
1 | 参考:工具+资源+提示注册 |
生产工具插件(shell、wekan、mattermost)可单独使用 PyPI包。看 mcp_tools.
代理功能
管理API
代理在单独的端口上运行FastAPI管理服务器(默认值:12299):
GET /proxy/status All plugins and tools
GET /proxy/plugins Plugin name list
POST /proxy/load {"plugin": "name"}
POST /proxy/unload {"plugin": "name"}
POST /proxy/reload {"plugin": "name"}
GET /proxy/commands Registered management extensions
POST /proxy/command/{name} Run a management extension
MCP管理工具
代理注册了任何连接的MCP客户端都可以调用的管理工具:
| 工具 | 说明 |
|---|---|
proxy__load |
在运行时加载插件 |
proxy__unload |
卸载插件 |
proxy__reload |
重新加载插件(获取代码更改) |
proxy__status |
显示加载的插件、工具、资源、提示 |
proxy__list |
列出可用(尚未加载)插件 |
proxy__tools |
列出所有加载的工具(支持 dynamic_only 过滤器) |
proxy__run |
按名称调用动态加载的工具(需要 dynamic_dispatch: true) |
服务器发送 tools/list_changed, resources/list_changed,以及 prompts/list_changed
插件加载或卸载时的通知。
管理API身份验证
可选择使用承载令牌保护管理API:
# proxy.yaml
management_token: "${MCP_MGMT_TOKEN}"
# Or via CLI
mcp-proxy serve --mgmt-token "my-secret" --autoload echo
# Client commands with token
mcp-proxy status --token "my-secret"
mcp-proxy load greet --token "my-secret"
# Or via environment variable (works for both server and client)
export MCP_MGMT_TOKEN="my-secret"
mcp-proxy serve --autoload echo
mcp-proxy status
如果没有配置令牌,则不需要身份验证(开发默认值)。
自动前缀
加载多个插件时避免工具名称冲突:
# proxy.yaml
auto_prefix: true # tool "send" from plugin "mm" becomes "mm_send"
plugins:
my_plugin:
prefix: "mm" # custom prefix (overrides plugin name)
echo:
prefix: false # disable prefix for this plugin
已经以插件前缀开头的工具不是双前缀的。
健康端点
在HTTP上运行时,健康服务器包括插件状态:
GET /health Simple status
GET /health/detailed Uptime, requests, errors
GET /health/ready Readiness check (verifies plugins loaded)
GET /health/plugins Current plugin and tool inventory
动态调度
MCP客户端缓存工具列表,大多数客户端不处理 tools/list_changed 通知。
这意味着在运行时加载的工具对客户端不可见。动态调度解决了这个问题
使用单个网关工具 proxy__run 它可以按名称调用任何动态加载的工具。
# proxy.yaml
dynamic_dispatch: true # default: false
autoload:
- echo # static — visible in tools/list, callable directly
# At runtime: load a plugin dynamically
proxy__load greet → "Loaded 'greet': 1 tool. Call them via proxy__run(...)"
# Call dynamic tools via the gateway
proxy__run(tool="greet", arguments={"name": "Claude"})
# List only dynamic tools
proxy__tools(dynamic_only=true)
安全规则: 启动时加载的工具(自动加载)有 不 可通过以下方式调用 proxy__run —
它们在 tools/list 客户端可以直接对其设置权限。
proxy__run 仅分派给启动后加载的工具。
注: 动态调度是当前MCP客户端生态系统的一种解决方法。 大多数客户端(Claude.ai、VS Code、LibreChat)不处理
tools/list_changed通知,因此运行时加载的工具对他们来说仍然是不可见的。如果客户端启动 本机支持动态工具发现,此功能可能会被弃用。
管理命令扩展
使用自定义管理命令扩展代理:
def register(mcp, config: dict) -> None:
proxy = config.get("_proxy")
if proxy:
proxy.register_command("stats", lambda p: f"{len(p.plugins)} plugins")
命令可通过MCP工具和REST管理API获得。
OAuth身份验证
OAuth是 默认启用 对于HTTP传输(对于stdio忽略)。
使用RFC 7662令牌自检——不需要额外的依赖关系(使用MCP SDK中的httpx)。
有效令牌缓存8小时(可通过配置 oauth_cache_ttl)以减少内省负载。
通过YAML或环境变量进行配置:
# config.yaml
oauth_enabled: true # default: true
oauth_server_url: "https://auth.example.com" # OAuth introspection endpoint
oauth_public_url: "https://mcp.example.com" # Public URL of this server
oauth_cache_ttl: 28800 # token cache in seconds (default: 8h)
# Or via environment
export MCP_OAUTH_ENABLED=true
export MCP_OAUTH_SERVER_URL=https://auth.example.com
export MCP_PUBLIC_URL=https://mcp.example.com
行为:
- 已配置HTTP+OAuth→ 通过内省进行完整令牌验证
- 启用了HTTP+OAuth,但缺少URL→ 警告,在没有身份验证的情况下运行
- HTTP+
oauth_enabled: false→ 无身份验证,无警告 - 标准输入输出→ 不管配置如何,OAuth都被忽略
要明确禁用:
oauth_enabled: false
插件注册表(v1.4)
该框架提供 PluginRegistry --内部使用的共享跟踪层
工厂和代理。如果你构建了一个自定义服务器,你可以直接使用它:
from mcp_server_framework.plugins import PluginRegistry
registry = PluginRegistry(mcp, config)
result = registry.load_plugin("my_plugin")
if result.ok:
print(f"Loaded: {result.plugin.tools}")
特征:
- 工具、资源和提示的碰撞检测
load_plugin()/unload_plugin()结构化结果(LoadResult,UnloadResult)get_summary(),all_tools,find_tool_owner()
内省助手
格式化输出的纯函数——无MCP依赖性:
from mcp_server_framework.plugins import plugin_status, plugin_list, tool_list
print(plugin_status(registry)) # Server info + counts
print(plugin_list(registry)) # Plugin → tools table
print(tool_list(registry)) # Sorted tool list
工具调用记录器
可插拔日志记录有四种实现方式:
from mcp_server_framework.plugins import (
JsonlToolLogger, # Daily JSONL + gzip archival + retention
TextToolLogger, # One-line text + size-based rotation
TranscriptLogger, # Markdown session transcripts
CompositeToolLogger, # Combines multiple loggers
)
logger = CompositeToolLogger([
JsonlToolLogger(log_dir),
TextToolLogger(log_file),
])
logger.log_call("echo", {"text": "hi"}, "hi", success=True)
安全
呼叫前验证
注册一个在每次工具调用之前运行的自定义验证器:
from mcp_server_framework.plugins import set_pre_call_validator
def my_validator(tool_name: str, params: dict) -> str | None:
"""Return error string to reject, None to allow."""
if len(str(params)) > 100_000:
return "Input too large"
return None
set_pre_call_validator(my_validator)
门——基于会话的TOTP保护(v1.5)
mcp_server_framework.gate 保护TOTP第二个因素背后的敏感工具组。所有组开始 锁定 每次重新启动代理时。有效的身份验证码会在可配置的不活动超时(默认值:2小时)内解锁组。任何工具调用都会重置计时器。
快速启动:
from mcp_server_framework.gate import Gate
def register(mcp, config):
gate = Gate(config.get("gate", {}))
gate.register_tools(mcp) # adds gate_unlock, gate_lock, gate_status
@mcp.tool()
@gate.protect("shell") # inner decorator — mcp.tool() must be outer
def shell_exec(command: str) -> str:
import subprocess
return subprocess.check_output(command, shell=True, text=True)
config.yaml:
gate:
enabled: true # false = tools work without authentication
secret_backend: env # env | file | vaultwarden
groups:
shell:
timeout: 7200 # inactivity timeout in seconds
secret_ref: MCP_GATE_SECRET_SHELL # env var name
vault:
timeout: 3600
secret_ref: MCP_GATE_SECRET_VAULT
MCP工具注册人 gate.register_tools(mcp):
| 工具 | 说明 |
|---|---|
gate_unlock(group, code) |
使用6位TOTP代码解锁组 |
gate_lock(group="") |
锁定一个组或所有组--终止开关 |
gate_status() |
显示每组的状态和剩余时间 |
秘密后端:
| 后端 | 配置 | 依赖关系 | 轮换 |
|---|---|---|---|
env |
secret_ref: ENV_VAR_NAME |
无 | 重新启动 |
file |
secret_ref: key, secret_file: ~/.mcp_gate_secrets |
无 | 立即 |
vaultwarden |
secret_ref: item-name, vaultwarden_url/token |
requests |
TTL之后 |
安全属性:连续5个错误代码后锁定,重启后没有持续的解锁状态,通过终止开关 gate_lock().
看 gate/README.md 获取完整文档。
日志记录
通过配置或环境配置日志格式:
log_level: INFO
log_format: json # "json" for machine-readable, "text" (default) for humans
JSON输出示例:
{"ts": "2026-03-11T10:30:00+00:00", "level": "INFO", "logger": "mcp_server_proxy.proxy", "msg": "Plugin 'echo' loaded: 2 tools ['echo', 'echo_upper']"}
工具调用日志记录(代理)
代理使用 JsonlToolLogger 从框架中记录每个工具调用 ~/.mcp_proxy/logs/tool_calls.jsonl:
{"ts": "2026-03-11T14:30:00", "tool": "echo", "params": {"text": "hello"}, "result": "hello", "ok": true}
- 采用gzip压缩的每日轮换(
YYYY-MM-DD.jsonl.gz) - 90天保留期,自动清理
- 安全——从不提高,从不阻止工具执行
交互式测试客户端
从终端调试和探索任何MCP服务器:
# Connect via stdio
python examples/mcp_client.py -v stdio -- mcp-proxy --autoload echo
# Connect via HTTP
python examples/mcp_client.py http http://localhost:12200/mcp
REPL命令: tools, call <name>, resources, prompts, info, quit
工具简写——直接按名称调用工具:
mcp> echo hello # single-arg shorthand
mcp> echo_upper world
mcp> greet Claude
mcp> proxy__status # no-arg tools work too
mcp> call echo # interactive mode (prompts for args)
示例
已准备好在中运行脚本 examples/:
./examples/run_factory_echo.sh # Factory + Echo (stdio, all-in-one)
./examples/run_proxy_http.sh # Proxy + Echo (HTTP)
./examples/run_proxy_full.sh # Proxy + Echo + Greet (HTTP)
./examples/connect_proxy_http.sh # Client → running proxy
看 示例/README.md 了解详情。
配置
带有环境变量覆盖的YAML配置(MCP_*):
server_name: "My Proxy"
transport: http # stdio | http
host: "0.0.0.0"
port: 12200
health_port: 12201
management_port: 12299
log_level: INFO
log_format: text # text | json
auto_prefix: true
dynamic_dispatch: false # true = enable proxy__run for runtime-loaded tools
# OAuth (enabled by default for HTTP, set false to disable)
oauth_server_url: "https://auth.example.com"
oauth_public_url: "https://mcp.example.com"
oauth_cache_ttl: 28800 # token cache seconds (default: 8h, 0 = disabled)
# Management API auth (separate from OAuth, optional)
management_token: "${MCP_MGMT_TOKEN}"
autoload:
- echo # local plugin
- mcp_shell_tools.shell # installed PyPI package
plugins:
echo:
enabled: true
插件特定配置
凭据和插件特定设置将进入单独的配置文件 每个插件, 不 在代理配置中:
~/mcp_plugins/{plugin_name}/config.yaml
代理会自动加载这些。如果不存在插件配置文件,它将下降
回到 plugins: 代理配置中的部分。
示例配置 config/:
| 配置 | 用例 |
|---|---|
proxy.example.yaml |
记录了所有选项的HTTP代理 |
config-stdio.yaml |
克劳德代码/克劳德桌面的stdio |
framework.example.yaml |
独立框架服务器 |
factory.example.yaml |
带有静态插件的工厂 |
克劳德代码/克劳德桌面
将代理用作stdio MCP服务器——无OAuth,无端口:
{
"mcpServers": {
"proxy": {
"command": "/path/to/mcp_server_toolkit/.venv/bin/mcp-proxy",
"args": ["serve", "--config", "config/config-stdio.yaml",
"--plugin-dir", "./plugins"]
}
}
}
测试
pytest # all tests
pytest -v # verbose
pytest tests/proxy/ # proxy only
项目结构
src/
├── mcp_server_framework/ # Shared: config, server, health, plugins
│ └── plugins/ # PluginRegistry, ToolLogger, Introspection
├── mcp_server_factory/ # Static loader + CLI
└── mcp_server_proxy/ # Dynamic loader + management API + CLI
plugins/
├── echo.py # Minimal example
├── greet.py # Minimal example
└── demo_full.py # Reference: tool + resource + prompt
examples/
├── mcp_client.py # Interactive test client
├── configs/ # Ready-to-use YAML configs
└── *.sh # Launch scripts
scripts/
└── new-plugin.sh # Plugin scaffold generator
tests/ # Tests (framework, factory, proxy)
config/ # Example configs + systemd service
生产工具插件在单独的存储库中维护: mcp_tools --壳牌、威肯、马特莫斯特等。
许可证
麻省理工学院