MCP生产演示——完整指南

该存储库演示了生产风格的MCP架构，包括：

stdio MCP服务器，
HTTP/SSE传输，
弹性MCP客户端，
高级模式（中间件、跟踪、启发、工具链），
生产挑战模式和测试，
一个统一的跑步者(main.py).

______________________________________________________________________

01_core_server.py --MCP服务器基础

这就是你的服务器所在的地方。它使用基于装饰器的工具进行注册 ToolRegistry 自动跟踪调用计数、错误率，并应用每个工具的超时。三种生产工具已全面实施： search_documents （带过滤器和分页）， execute_query （具有注入防止的SQL——阻止DROP、DELETE等），以及 send_notification （松弛/电子邮件/传呼机）。A. ResourceProvider 将实时健康检查和工具模式作为MCP资源公开。提示模板 analyze_query_results 和 incident_response 包括在内。JSON结构化日志封装了一切。

02_http_transport.py --网络就绪的HTTP+SSE传输

当您需要多客户端、云托管的MCP（而不仅仅是stdio）时，这是您的层。基于FastAPI和JWT身份验证中间件构建，租户上下文隔离（每个请求都携带一个 TenantContext 具有权限和级别），以及 SSEConnectionManager 对于推送活动。包括安全标头中间件（HSTS、XSS保护）、 /token 端点，主要 /mcp JSON-RPC端点，以及 /mcp/events 对于SSE订阅。通过适当的保活清理实现优雅的关机。

03_client.py --具有弹性的生产客户

MCP在生产中最难的部分是客户。这实现了一个完整的 CircuitBreaker （关闭→ OPEN → HALF_OPEN状态机）、带抖动的指数退避重试、幂等工具的TTL响应缓存，以及 call_tools_concurrent() 用于有界并行执行 asyncio.Semaphore上交所 subscribe_events() 是一个异步生成器，在事件到达时生成事件。HTTP/2和连接池已在 httpx.AsyncClient.

04_advanced_concepts.py --八种先进的MCP模式

中间件管道在每个工具调用周围运行pre/post钩子，在返回过程中按顺序前进和后退——就像Python一样 contextmanager 堆叠。 SamplingAPI 显示了服务器如何从客户端中间工具请求LLM推理。 ToolChain 通过条件跳过步骤和在步骤之间传递上下文，实现多步骤编排。带有验证器的Pydantic模型强制执行结构化输出契约。OpenTetry跨越了包装工具的执行。这 RootsManager 处理工作区文件系统暴露。 ElicitationRequest 允许工具在执行过程中要求用户提供更多输入。 DynamicToolLoader 在运行时从config加载工具，而无需重新启动。

05_challenges.py --生产中断的原因以及如何修复

协议版本协商可以优雅地处理旧客户端。 ToolVersionRegistry 使用参数迁移管理模式演化（旧 {"q":"...", "max":10} → new {"query":"...", "limit":10}). MCPError 将所有错误分类为具有恢复策略的类型化代码（速率受限→ 指数回退，找不到资源→ 缓存回退，断路→ 优雅地降级）。 ConnectionManager 处理挂起请求清理的重新连接。 SecretManager 在缓存接口后抽象env-vars/AWS Secrets Manager/Vault。测试套件涵盖了断路器状态机、中间件执行顺序、SQL注入、速率限制、参数迁移和工具链条件。

1）快速入门

先决条件

Python 3.11+（项目已在Python 3.13上验证）
pip

安装依赖项

pip install -r requirements.txt

环境设置(`.env`)

SLACK_API_TOKEN=your-slack-token

笔记：

已实施Slack通知。
在当前实现中，电子邮件通道被故意禁用。

______________________________________________________________________

2）如何运行一切（统一入口点）

使用 main.py 对于所有常见任务：

python main.py --help
python main.py all
python main.py check
python main.py http
python main.py core
python main.py client-demo
python main.py test

命令行为

all：导入检查所有模块，检查HTTP运行状况（如果服务器正在运行），运行测试。
check：导入检查所有编号的模块文件。
http：从运行FastAPI HTTP+SSE服务器 02_http_transport.py.
core：从以下位置运行stdio MCP服务器 01_core_server.py.
client-demo：从运行演示客户端流 03_client.py （需要运行HTTP服务器）。
test：run pytest 05_challenges.py -v --tb=short.

______________________________________________________________________

3）高层架构

01_core_server.py：stdio上的MCP服务器逻辑（工具/资源/提示）。
02_http_transport.py：基于HTTP+SSE的JSON-RPC推送传输、JWT身份验证、租赁。
03_client.py：强大的MCP客户端（重试、断路器、缓存、并发调用）。
04_advanced_concepts.py：现实世界系统的高级实施模式。
05_challenges.py：生产挑战解决方案+集成测试。
main.py：统一的操作CLI。

______________________________________________________________________

4）逐个文件，逐个函数引用

`01_core_server.py` --核心MCP服务器（stdio传输）

目的

使用工具注册、资源、提示、日志记录和安全控制来实现主MCP服务器。

类和方法

`JSONFormatter`

format(record)：发出带有时间戳、级别、消息、模块、函数和可选异常/额外字段的结构化JSON日志行。

用例：用于可观察性管道的机器可解析日志记录。

`RateLimiter`

is_allowed(client_id)：令牌桶式窗口检查；回报 (allowed, metadata) 包括剩余配额或之后重试。

用例：保护昂贵的工具免受滥用/尖峰。

`ToolMetadata` （数据类）

按工具合约保存：名称、描述、JSON模式、处理程序、超时、标签、认证/速率选项。

`ToolRegistry`

__init__()：初始化工具注册表和调用/错误度量。
register(...)：decorator工厂，注册一个工具并用超时、调用/错误计数和结构化日志对其进行包装。

- 嵌套 decorator(func)：绑定元数据并返回包装器。 - 嵌套 wrapper(*args, **kwargs)：具有超时/错误处理功能的运行时执行包装器。

get_mcp_tools()：将注册表元数据转换为MCP types.Tool 物体。
stats()：返回每次工具调用/错误/错误率摘要。

用例：具有指标+弹性的标准化工具生命周期。

`ResourceProvider`

get_resource(uri)：提供虚拟资源(config://server/info, config://server/health, schema://tools).
_run_health_checks()：执行健康探测（模拟数据库+外部HTTP检查），包括工具统计信息。

用例：将内部服务器状态作为可读资源公开给MCP客户端。

顶级工具功能

`setup_logging(level="INFO")`

配置 mcp.server logger发出JSON日志。

`search_documents(query, collection="all", limit=10, filters=None)`

模拟全文/语义搜索，返回排名的文档。

用例：检索样式MCP工具。

`execute_query(sql, database="analytics", timeout=30, max_rows=1000)`

带防护栏的只读SQL执行：

仅允许 SELECT
阻止危险的SQL关键字(DROP, DELETE等等）

用例：安全分析查询工具。

`send_notification(channel, destination, subject, body, priority="normal", metadata=None)`

具有特定通道行为的通知调度：

Slack频道发送 chat.postMessage 使用 SLACK_API_TOKEN.
电子邮件路径故意提高 NotImplementedError.
PagerDuty/webhook是占位符/无操作警告路径。

用例：警报/事件通信工具。

`build_prompt_messages(name, args)`

为以下对象构建结构化提示消息：

analyze_query_results
incident_response

用例：服务器提供可重用的提示模板。

`create_server()`

构建MCP Server 并注册处理程序：

嵌套 list_tools()
嵌套 call_tool(name, arguments)
嵌套 list_resources()
嵌套 read_resource(uri)
嵌套 list_prompts()
嵌套 get_prompt(name, arguments)

用例：一个地方连接所有MCP功能。

`main()`

使用启动stdio MCP服务器 stdio_server() 上下文和初始化选项。

______________________________________________________________________

`02_http_transport.py` --HTTP+SSE MCP传输

目的

使用JWT身份验证、租赁上下文、JSON-RPC端点和SSE事件将MCP功能封装在FastAPI后面。

类和方法

`ServerConfig` （数据类）

中央服务器配置（主机/端口/JWT/CORS/请求限制/SSE定时）。

`TenantContext` （数据类）

can(permission)：权限检查(admin 旁路）。
rate_limit()：层感知请求限制(free/pro/enterprise).

用例：按租户授权和策略路由。

`MCPRequest` / `MCPResponse` （Pydantic模型）

键入JSON-RPC请求/响应模型。

`SSEConnectionManager`

__init__()：初始化每个租户队列注册表。
connect(tenant_id)：为一个SSE客户端分配队列。
disconnect(tenant_id, queue)：删除队列。
broadcast(tenant_id, event)：将活动分为一个租户。
broadcast_all(event)：向所有租户推广活动。
connection_count （属性）：总活动SSE流。

用例：可扩展的服务器推送通知。

`ToolDispatcher`

dispatch(method, params, ctx)：中央JSON-RPC方法路由器。
_handle_initialize(...)：协议握手+功能。
_handle_list_tools(...)：权限筛选工具列表。
_handle_call_tool(...)：权限检查、模拟执行、事件广播。
_handle_list_resources(...)：资源列表响应。
_handle_read_resource(...)：使用租户隔离读取资源。
_handle_list_prompts(...)：提示列表。
_handle_get_prompt(...)：迅速实现。

用例：传输层JSON-RPC控制器。

顶级功能/路线

`create_token(tenant_id, user_id, permissions, tier)`

使用过期和JTI签署JWT访问令牌。

`require_auth(request, credentials)`

受保护端点的JWT验证依赖关系。将租户+关联ID附加到请求状态。

`lifespan(app)`

FastAPI生命周期管理器：

启动日志，
背景直播，
优雅关机。

嵌套助手：

keepalive()：发出周期性ping事件。

`security_headers(request, call_next)`

应用安全标头并传播相关ID的中间件。

路由处理程序

health()：未经身份验证的活性端点。
create_access_token(request)：从请求正文字段发出JWT。
mcp_endpoint(request, ctx)：主MCP JSON-RPC端点。
sse_events(request, ctx)：事件流终结点。

- 嵌套 event_generator()：发出初始连接事件+租户事件+keepalive注释。

server_stats(ctx)：仅限管理员的操作统计端点。

______________________________________________________________________

`03_client.py` --生产MCP客户端

目的

提供具有重试、断路器、缓存、并发控制和SSE消耗的强大异步MCP客户端。

类和方法

`CircuitState` （枚举）

CLOSED, OPEN, HALF_OPEN.

`CircuitBreaker`

record_success()：更新状态，并在足够多的半开成功后关闭断路器。
record_failure()：跟踪滚动故障，在阈值处打开断路器。
can_request()：看门人；根据恢复超时打开/半打开。
state （属性）：断路器状态。
stats （属性）：诊断计数器。

用例：防止下游停机期间发生级联故障。

`RetryConfig`

delay_for_attempt(attempt)：指数退避+可选抖动。
should_retry(status_code, exc)：状态/异常重试策略。

用例：稳健的瞬态故障处理。

`ResponseCache`

__init__(maxsize, ttl)：初始化TTL缓存。
_key(tool, args)：用于工具输入的稳定哈希键。
get(tool, args)：幂等工具的缓存查找。
set(tool, args, value)：可缓存工具的缓存插入。
info （属性）：缓存诊断。

用例：减少重复的只读工具负载。

`MCPClient`

__init__(...)：配置客户端传输、重试、中断、缓存、关联ID。
__aenter__()：打开HTTP客户端并运行MCP初始化握手。
__aexit__()：关闭HTTP连接。
_next_id()：JSON-RPC ID生成器。
_initialize()：发送 initialize 方法并标记客户端就绪。
_rpc(method, params)：具有重试+断路器行为的核心JSON-RPC调用路径。
list_tools()：呼叫 tools/list.
call_tool(name, arguments)：缓存感知工具调用+内容解析。
call_tools_concurrent(calls, max_concurrency=5)：有界并行调用。

- 嵌套 bounded_call(name, args)：每次调用的信号量包装器。

list_resources()：呼叫 resources/list.
read_resource(uri)：呼叫 resources/read 并提取文本。
list_prompts()：呼叫 prompts/list.
get_prompt(name, arguments=None)：呼叫 prompts/get.
subscribe_events()：异步SSE消费者生成JSON事件。
stats （属性）：聚合断路器/缓存/会话诊断。

顶级函数

`get_token(base_url, tenant_id, user_id)`

请求JWT的便利方法 /token.

`demo_client()`

演示完整的客户端生命周期：

auth，
工具发现，
单次+并发工具调用，
资源/提示使用，
SSE事件订阅。

______________________________________________________________________

`04_advanced_concepts.py` --高级MCP模式

目的

展示生产MCP系统中使用的高级架构模式。

1）中间件管道

`ToolCallContext` （数据类）

用于中间件/工具执行的共享可变请求上下文。

`Middleware` （摘要）

before(ctx) / after(ctx)：生命周期挂钩。

`MiddlewarePipeline`

__init__()：保存中间件列表。
use(middleware)：附加中间件。
run(ctx, handler)：在钩子、处理程序之前执行，然后在钩子之后反向执行。

混凝土中间件

AuditLogMiddleware.before/after：审核日志记录，并对正文/PII大小字段进行编辑。
PiiRedactionMiddleware.before/after/_redact：递归屏蔽已知的敏感密钥。
CostTrackingMiddleware.__init__/before/after/get_usage：按租户估算每个呼叫令牌的成本。
ArgumentValidationMiddleware.__init__/before/after：根据JSON模式验证输入参数。

2） API取样

`SamplingClient`

__init__(session)：存储MCP会话。
create_message(...)：要求客户端模型进行推理/分析。
analyze_with_llm(data, task)：结构化分析提示的助手。

3）工具链

`ChainStep` （数据类）

使用工具名称、输入映射器、可选条件、输出键和错误行为停止定义一个步骤。

`ToolChain`

__init__(steps)：商店订购连锁步骤。
run(initial_context, tool_executor)：通过上下文传递执行条件多步编排。

4）结构化输出验证

模型

DocumentResult
SearchOutput 带验证器 sort_by_score(...)
IncidentReport

效用

validate_tool_output(model_class, raw_output)：验证并返回类型化模型。

5）遥测和追踪

`MCPTelemetry`

__init__(service_name)：设置服务元数据和span store。
start_span(name, attributes=None)：开始逻辑跨度。
end_span(span, error=None)：关闭跨度并标记状态。
add_event(span, name, attributes=None)：将事件附加到span。
export_spans()：返回收集的跨度。

`with_tracing(tool_name)`

在遥测生命周期中封装异步工具的装饰工厂。

嵌套 decorator(func)
嵌套 wrapper(*args, **kwargs)

6）根协议

`RootsManager`

__init__()：初始化根列表和侦听器。
add_root(uri, name)：添加root并通知侦听器。
remove_root(uri)：删除根并通知侦听器。
on_change(handler)：订阅侦听器。
roots （property）：返回根列表。
resolve_path(relative_path)：解析第一个文件根的相对路径。

7）激励

`ElicitationField` （数据类）

用户输入请求的字段定义。

`ElicitationRequest`

__init__(session)：存储会话。
ask(prompt, fields, title="Input Required")：通过MCP启发流请求结构化用户输入。
_field_to_schema(field)：将字段模型转换为JSON模式片段。

8）动态刀具加载

`DynamicToolLoader`

__init__(server)：保存服务器和加载的工具映射。
load_from_config(config)：在运行时从配置映射创建工具。
call_dynamic_tool(name, arguments)：执行加载的工具函数。
unload_tool(name)：删除加载的工具。
loaded_tools （属性）：列出当前加载的动态工具。

高级服务器组成

`AdvancedMCPServer`

__init__()：组装服务器、管道、根、启发、动态工具。
_setup_handlers()：注册MCP处理程序。

- 嵌套 call_tool(name, arguments) - 嵌套 handler(**kwargs) 用于流水线执行 - 嵌套 list_tools()

run_incident_workflow(incident_title, severity, symptoms)：演示启发+工具链。
tool_executor(tool_name, args)：工作流用于执行链步骤的助手。

______________________________________________________________________

`05_challenges.py` --生产挑战+测试

目的

记录常见的故障模式和处理它们的实用模式，然后用pytest验证它们。

公用事业装载机

`_load_module(filename, module_name)`

加载数字前缀模块（例如。， 01_core_server.py)via importlib.

用例：避免测试/示例中的Python标识符限制。

1）版本控制和模式演变

`ProtocolVersion`

negotiate(client_version)：查找支持的最佳协议版本。
feature_supported(version, feature)：按版本进行特征门控。

`ToolVersionRegistry`

__init__()：初始化架构/版本存储。
register_version(...)：注册工具模式版本+可选弃用通知。
get_schema(tool_name, client_version)：选择最高兼容架构。
migrate_arguments(tool, args, from_ver, to_ver)：将旧有效载荷转换为新格式。

2）键入和恢复时出错

`MCPErrorCode` （枚举）

JSON-RPC+MCP特定错误代码。

`MCPError`

to_dict()：JSON-RPC样式的错误形状。
from_exception(exc)：将通用异常映射到键入的MCP错误。

`ErrorRecoveryManager`

__init__(alert_webhook=None)：设置策略状态。
handle(error, context)：应用策略（回退/回退/降级/警报）。
_send_alert(error, context)：可选的webhook警报。

3）连接生命周期

`ConnectionState` （枚举）

客户端/服务器连接生命周期的状态机。

`ConnectionManager`

__init__(reconnect_attempts=5, reconnect_delay=2.0)
on_state_change(state, handler)
_transition(new_state)
connect(connector)
reconnect(connector)
is_active （财产）
stats （财产）

4）秘密管理

`SecretManager`

__init__(provider="env")
get(key)
_fetch(key)
load_server_config()

用例：具有提供者抽象的中央、缓存支持的秘密检索。

5）集成测试线束

`MCPTestServer`

__init__(server)
call_tool(name, arguments)
recorded_calls （财产）
assert_tool_called(name, times=1)
assert_tool_called_with(name, **expected_args)

Pytest夹具和测试

夹具

mock_tool_registry() 使用嵌套异步模拟：

- mock_search(**kwargs) - mock_query(**kwargs)

测试

test_protocol_version_negotiation()
test_argument_migration()
test_sql_injection_prevention()
test_rate_limiter_enforcement()
test_circuit_breaker_state_machine()
test_middleware_pipeline_execution_order()

- 包括嵌套 LoggingMiddleware.__init__/before/after - 包括嵌套 handler(**kwargs)

test_tool_chain_condition_skip()

- 包括嵌套 executor(tool_name, args)

______________________________________________________________________

`main.py` --统一运行程序（操作CLI）

目的

提供一个命令行入口点来运行/检查/测试整个存储库。

函数

`load_module(filename, module_name)`

用于编号文件名的动态导入助手。

`run_core_server()`

通过启动stdio MCP服务器 01_core_server.main().

`run_http_server()`

从启动FastAPI/Uvicorn应用程序 02_http_transport 配置。

`run_client_demo()`

跑 03_client.demo_client().

`run_tests()`

对执行pytest 05_challenges.py 并返回退出代码。

`check_all_imports()`

加载所有主要模块并打印成功行。

`check_http_health(url, timeout)`

简单的HTTP健康探测 /health.

`run_all_flow()`

复合健全流：导入、健康探测、测试。

`build_parser()`

定义CLI参数和命令选项。

`main()`

入口调度和进程出口行为。

______________________________________________________________________

5）典型开发流程

A）快速验证项目

python main.py all

B）运行服务器+客户端演示

1号航站楼：

python main.py http

2号航站楼：

python main.py client-demo

C）运行测试套件

python main.py test

______________________________________________________________________

6）操作说明

如果 python main.py http 端口正在使用时发生故障 8080，停止上一个进程或更改端口 ServerConfig.
保守秘密 .env；永远不要承诺真正的代币。
结构化日志是JSON格式，适用于日志管理系统。
SQL工具故意是只读的，并阻止破坏性关键字。

______________________________________________________________________

7）当前通知范围

现已实施：

Slack通过 send_notification(channel="slack", ...) 使用 SLACK_API_TOKEN.

目前尚未实施：

电子邮件（明确 NotImplementedError)
PagerDuty/Webhook完全集成（占位符路径）

______________________________________________________________________

MCP prod demo

快速接入

详细介绍

MCP生产演示——完整指南

1） 快速入门

先决条件

安装依赖项

环境设置(.env)

2） 如何运行一切（统一入口点）

命令行为

3） 高层架构

4） 逐个文件，逐个函数引用

01_core_server.py --核心MCP服务器（stdio传输）

目的

类和方法

JSONFormatter

RateLimiter

ToolMetadata （数据类）

ToolRegistry

ResourceProvider

顶级工具功能

setup_logging(level="INFO")

search_documents(query, collection="all", limit=10, filters=None)

execute_query(sql, database="analytics", timeout=30, max_rows=1000)

send_notification(channel, destination, subject, body, priority="normal", metadata=None)

build_prompt_messages(name, args)

create_server()

main()

02_http_transport.py --HTTP+SSE MCP传输

目的

类和方法

ServerConfig （数据类）

TenantContext （数据类）

MCPRequest / MCPResponse （Pydantic模型）

SSEConnectionManager

ToolDispatcher

顶级功能/路线

create_token(tenant_id, user_id, permissions, tier)

require_auth(request, credentials)

lifespan(app)

security_headers(request, call_next)

路由处理程序

03_client.py --生产MCP客户端

目的

类和方法

CircuitState （枚举）

CircuitBreaker

RetryConfig

ResponseCache

MCPClient

顶级函数

get_token(base_url, tenant_id, user_id)

demo_client()

04_advanced_concepts.py --高级MCP模式

目的

1） 中间件管道

ToolCallContext （数据类）

Middleware （摘要）

MiddlewarePipeline

混凝土中间件

2） API取样

SamplingClient

3） 工具链

ChainStep （数据类）

ToolChain

4） 结构化输出验证

模型

效用

5） 遥测和追踪

MCPTelemetry

with_tracing(tool_name)

6） 根协议

RootsManager

7） 激励

ElicitationField （数据类）

ElicitationRequest

8） 动态刀具加载

DynamicToolLoader

高级服务器组成

AdvancedMCPServer

1）快速入门

环境设置(`.env`)

2）如何运行一切（统一入口点）

3）高层架构

4）逐个文件，逐个函数引用

`01_core_server.py` --核心MCP服务器（stdio传输）

`JSONFormatter`

`RateLimiter`

`ToolMetadata` （数据类）

`ToolRegistry`

`ResourceProvider`

`setup_logging(level="INFO")`

`search_documents(query, collection="all", limit=10, filters=None)`

`execute_query(sql, database="analytics", timeout=30, max_rows=1000)`

`send_notification(channel, destination, subject, body, priority="normal", metadata=None)`

`build_prompt_messages(name, args)`

`create_server()`

`main()`

`02_http_transport.py` --HTTP+SSE MCP传输

`ServerConfig` （数据类）

`TenantContext` （数据类）

`MCPRequest` / `MCPResponse` （Pydantic模型）

`SSEConnectionManager`

`ToolDispatcher`

`create_token(tenant_id, user_id, permissions, tier)`

`require_auth(request, credentials)`

`lifespan(app)`

`security_headers(request, call_next)`

`03_client.py` --生产MCP客户端

`CircuitState` （枚举）

`CircuitBreaker`

`RetryConfig`

`ResponseCache`

`MCPClient`

`get_token(base_url, tenant_id, user_id)`

`demo_client()`

`04_advanced_concepts.py` --高级MCP模式

1）中间件管道

`ToolCallContext` （数据类）

`Middleware` （摘要）

`MiddlewarePipeline`

`SamplingClient`

3）工具链

`ChainStep` （数据类）

`ToolChain`

4）结构化输出验证

5）遥测和追踪

`MCPTelemetry`

`with_tracing(tool_name)`

6）根协议

`RootsManager`

7）激励

`ElicitationField` （数据类）

`ElicitationRequest`

8）动态刀具加载

`DynamicToolLoader`

`AdvancedMCPServer`

`05_challenges.py` --生产挑战+测试

`_load_module(filename, module_name)`

1）版本控制和模式演变

`ProtocolVersion`

`ToolVersionRegistry`

2）键入和恢复时出错

`MCPErrorCode` （枚举）

`MCPError`

`ErrorRecoveryManager`

3）连接生命周期

`ConnectionState` （枚举）

`ConnectionManager`

4）秘密管理

`SecretManager`

5）集成测试线束

`MCPTestServer`

`main.py` --统一运行程序（操作CLI）

`load_module(filename, module_name)`

`run_core_server()`

`run_http_server()`

`run_client_demo()`

`run_tests()`

`check_all_imports()`