MCP服务器框架

一个用于构建MCP(模型上下文协议)服务器的Python框架,支持插件开发和动态管理。

  • by cuber-it
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 cuber-it/mcp_server_toolkit

使用说明

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 --壳牌、威肯、马特莫斯特等。

许可证

麻省理工学院