MCP共享
一个Python库，为构建具有较少样板和一致模式的模型上下文协议（MCP）服务器提供可重用的基础设施。
概述
MCP Commons为构建可维护的MCP服务器提供了架构模式：
主要价值（90%）:
- 适配器模式 -将业务逻辑与MCP协议解耦，以实现多接口重用
- 用例结果 -所有操作中一致的错误处理模式
便利功能（10%）:
- 批量操作 -带有错误报告的配置驱动工具注册
- 工具生命周期 -通过FastMCP的add_tool（）和remove_tool（
基于FastMCP构建：mcp-commons是FastMCP现有方法的精简封装。它不会取代FastMCP的功能——它提供了架构模式和便利的包装器，使您的代码更易于维护。
______________________________________________________________________
为什么存在mcp公共资源
装饰师的问题
MCP SDK使用装饰器(@server.tool())将功能注册为工具。虽然使功能暴露变得容易的目标令人钦佩， 装饰者是错误的机制 为了这个目的。
装饰人员应添加跨领域的关注点 （缓存、身份验证、日志记录），无论函数如何使用，都适用。它们不应该指定使用上下文（MCP vs REST vs CLI）。
# ❌ PROBLEM: Function is now ONLY usable in MCP context
from mcp.server.fastmcp import FastMCP
server = FastMCP("my-server")
@server.tool()
async def search_documents(query: str) -> dict:
"""This function is tied to MCP - can't reuse for REST, CLI, or testing."""
results = await document_service.search(query)
return {"results": results}
# Can't use this function in:
# - REST API endpoints
# - CLI commands
# - GraphQL resolvers
# - Unit tests (without MCP context)解决方案：适配器模式
适配器/包装器功能 提供相同的易用性，同时保持适当的关注点分离。您的业务逻辑保持纯粹，与框架无关，而瘦适配器处理协议转换。
# ✅ SOLUTION: Pure business logic, reusable everywhere
async def search_documents(query: str) -> List[Document]:
"""Pure function - no MCP coupling, works anywhere."""
return await document_service.search(query)
# MCP adapter - thin wrapper for protocol translation
@server.tool()
async def mcp_search(query: str) -> dict:
results = await search_documents(query)
return {"results": [doc.to_dict() for doc in results]}
# REST API - reuses same logic
@app.get("/api/search")
async def api_search(query: str):
results = await search_documents(query)
return {"results": [doc.to_dict() for doc in results]}
# CLI - reuses same logic
@cli.command()
def cli_search(query: str):
results = asyncio.run(search_documents(query))
for doc in results:
print(f"- {doc.title}")
# Testing - pure function, no framework needed
async def test_search():
results = await search_documents("test query")
assert len(results) > 0建筑效益
此适配器模式支持：
- 干燥原理 -一个业务功能，多个接口
- 关注点分离 -独立于运输的业务逻辑
- 框架独立性 -不与MCP SDK、FastAPI、Click等耦合。
- 易于测试 -测试没有框架上下文的纯函数
- 面向未来 -当MCP SDK v2.0更改时，只有适配器需要更新
mcp-commons之所以存在，是因为mcp SDK在这个基本的设计决策上犯了错误。 适配器模式并不“好用”，它对于任何非平凡应用程序中的正确架构都是必不可少的。
______________________________________________________________________
目录
- 工具适配器 - 批量注册 - 工具管理（v1.2.0）
______________________________________________________________________
安装
需求
- python：3.11+（推荐3.13）
- MCP-SDK: 1.27.0+
- 依赖项：Pydantic 2.12.5+，PyYAML 6.0.3+
从PyPI安装
pip install mcp-commons安装用于开发
git clone https://github.com/dawsonlp/mcp-commons.git
cd mcp-commons
pip install -e ".[dev]"______________________________________________________________________
FastMCP提供什么与mcp commons添加什么
| 功能 | FastMCP（SDK） | mcp-commons |
|---|---|---|
| 工具注册 | server.add_tool(func, name, desc) | 配置驱动的批量包装器 |
| 工具拆卸 | server.remove_tool(name) （v1.17.0+） | 带报告的批量包装器 |
| 装饰器 | @server.tool() 装饰师 | ❌ 我们不使用装饰师 |
| 适配器模式 | ❌ 未提供 | ✅ 核心功能-解耦逻辑 |
| 用例结果 | ❌ 未提供 | ✅ 一致的错误处理 |
| 错误报告 | 失败异常 | 成功/失败批处理报告 |
| 工具管理器 | 工具管理器、资源管理器等。 | ✅ 我们使用FastMCP的经理 |
关键点：mcp-commons不会取代FastMCP，它建立在它之上。我们使用FastMCP的 add_tool() 和 remove_tool() 方法内部，在上面添加方便的包装器和架构模式。
______________________________________________________________________
快速开始
1.基本适配器模式
将异步函数转换为MCP工具：
from mcp_commons import create_mcp_adapter, UseCaseResult
from mcp.server.fastmcp import FastMCP
# Create MCP server
server = FastMCP("my-server")
# Your business logic
async def search_documents(query: str, limit: int = 10) -> UseCaseResult:
"""Search documents with natural language query."""
results = await document_service.search(query, limit)
return UseCaseResult.success_with_data({
"results": results,
"count": len(results)
})
# Register as MCP tool (adapter handles conversion automatically)
@server.tool()
async def search(query: str, limit: int = 10) -> dict:
adapter = create_mcp_adapter(search_documents)
return await adapter(query=query, limit=limit)2.批量注册
一次注册多个工具：
from mcp_commons import bulk_register_tools
# Define tool configurations
tools_config = {
"list_projects": {
"function": list_projects_handler,
"description": "List all projects"
},
"create_project": {
"function": create_project_handler,
"description": "Create a new project"
},
"delete_project": {
"function": delete_project_handler,
"description": "Delete a project by ID"
}
}
# Register all at once with consistent error handling
registered = bulk_register_tools(server, tools_config)
print(f"Registered {len(registered)} tools")3.工具管理（v1.2.0）
在运行时动态管理工具：
from mcp_commons import (
bulk_remove_tools,
bulk_replace_tools,
get_registered_tools,
tool_exists
)
# Check what tools exist
all_tools = get_registered_tools(server)
print(f"Currently registered: {all_tools}")
# Remove deprecated tools
result = bulk_remove_tools(server, ["old_tool1", "old_tool2"])
print(f"Removed {len(result['removed'])} tools")
# Hot-reload: replace tools atomically
result = bulk_replace_tools(
server,
tools_to_remove=["v1_search"],
tools_to_add={
"v2_search": {
"function": improved_search,
"description": "Enhanced search with filters"
}
}
)______________________________________________________________________
核心功能
工具适配器
适配器模式自动处理业务逻辑和MCP工具格式之间的转换。
基本用法
from mcp_commons import create_mcp_adapter, UseCaseResult
async def calculate_metrics(dataset_id: str) -> UseCaseResult:
"""Calculate metrics for a dataset."""
try:
data = await load_dataset(dataset_id)
metrics = compute_metrics(data)
return UseCaseResult.success_with_data(metrics)
except DatasetNotFoundError as e:
return UseCaseResult.failure(f"Dataset not found: {e}")
except Exception as e:
return UseCaseResult.failure(f"Calculation failed: {e}")
# Create adapter
adapted = create_mcp_adapter(calculate_metrics)
# Use in MCP server
@server.tool()
async def metrics(dataset_id: str) -> dict:
return await adapted(dataset_id=dataset_id)错误处理
适配器提供一致的错误响应：
# Success response
UseCaseResult.success_with_data({"status": "completed", "value": 42})
# Returns: {"success": True, "data": {...}, "error": None}
# Failure response
UseCaseResult.failure("Invalid input parameters")
# Returns: {"success": False, "data": None, "error": "Invalid input parameters"}批量注册
FastMCP的便利包装 add_tool() 配置驱动注册的方法：
它实际上做了什么:
# mcp-commons bulk_register_tools() is essentially:
for tool_name, config in tools_config.items():
server.add_tool( # ← FastMCP's existing method
config["function"],
name=tool_name,
description=config["description"]
)
# Plus: error handling, logging, and success/failure reporting为什么要使用它：配置驱动的API+批处理错误处理，而不是手动循环。
配置字典
tools_config = {
"tool_name": {
"function": async_function,
"description": "Tool description",
# Optional metadata
}
}
registered = bulk_register_tools(server, tools_config)元组格式（简单）
from mcp_commons import bulk_register_tuple_format
tools = [
("list_items", list_items_function),
("get_item", get_item_function),
("create_item", create_item_function),
]
bulk_register_tuple_format(server, tools)使用适配器模式
from mcp_commons import bulk_register_with_adapter_pattern
# All functions return UseCaseResult
use_cases = {
"validate_data": validate_data_use_case,
"process_data": process_data_use_case,
"export_data": export_data_use_case,
}
bulk_register_with_adapter_pattern(
server,
use_cases,
adapter_function=create_mcp_adapter
)工具管理（v1.2.0）
1.2.0版本中的新功能：用于批处理工具操作的便利包装器。
它实际上做了什么：FastMCP上的循环 remove_tool() 方法（在SDK v1.17.0中添加），带有错误报告：
# mcp-commons bulk_remove_tools() is essentially:
for tool_name in tool_names:
try:
server.remove_tool(tool_name) # ← FastMCP's method (v1.17.0+)
removed.append(tool_name)
except Exception as e:
failed.append((tool_name, str(e)))
# Returns: {"removed": [...], "failed": [...], "success_rate": 66.7}为什么要使用它：批量操作+详细的成功/失败报告，而不是手动循环。
删除工具
from mcp_commons import bulk_remove_tools
# Remove multiple tools
result = bulk_remove_tools(server, ["deprecated_tool1", "deprecated_tool2"])
# Check results
print(f"Removed: {result['removed']}")
print(f"Failed: {result['failed']}")
print(f"Success rate: {result['success_rate']:.1f}%")更换工具（热重新加载）
from mcp_commons import bulk_replace_tools
# Atomically swap old tools for new ones
result = bulk_replace_tools(
server,
tools_to_remove=["old_search", "old_filter"],
tools_to_add={
"new_search": {
"function": enhanced_search,
"description": "Improved search with AI"
},
"new_filter": {
"function": enhanced_filter,
"description": "Advanced filtering"
}
}
)有条件移除
from mcp_commons import conditional_remove_tools
# Remove tools matching a pattern
removed = conditional_remove_tools(
server,
lambda name: name.startswith("test_") or "deprecated" in name.lower()
)
print(f"Cleaned up {len(removed)} tools")工具检查
from mcp_commons import get_registered_tools, tool_exists, count_tools
# List all tools
tools = get_registered_tools(server)
print(f"Available tools: {tools}")
# Check specific tool
if tool_exists(server, "search_documents"):
print("Search tool is available")
# Get count
total = count_tools(server)
print(f"Total tools registered: {total}")______________________________________________________________________
高级用法
自定义错误处理程序
from mcp_commons import create_mcp_adapter
def custom_success_handler(result):
"""Custom formatting for successful results."""
return {
"status": "success",
"payload": result.data,
"timestamp": datetime.now().isoformat()
}
def custom_error_handler(result):
"""Custom formatting for errors."""
return {
"status": "error",
"message": result.error,
"timestamp": datetime.now().isoformat()
}
adapted = create_mcp_adapter(
my_function,
success_handler=custom_success_handler,
error_handler=custom_error_handler
)验证和记录
from mcp_commons import validate_tools_config, log_registration_summary
# Validate before registering
try:
validate_tools_config(tools_config)
except ValueError as e:
print(f"Invalid configuration: {e}")
# Register with logging
registered = bulk_register_tools(server, tools_config)
log_registration_summary(registered, len(tools_config), "MyServer")测试您的工具
import pytest
from mcp_commons import create_mcp_adapter, UseCaseResult
@pytest.mark.asyncio
async def test_search_tool():
"""Test search tool with adapter."""
async def mock_search(query: str) -> UseCaseResult:
return UseCaseResult.success_with_data({"results": ["doc1", "doc2"]})
adapted = create_mcp_adapter(mock_search)
result = await adapted(query="test")
assert result["success"] is True
assert len(result["data"]["results"]) == 2______________________________________________________________________
API 参考
核心功能
create_mcp_adapter()
将异步函数转换为MCP兼容的工具适配器。
参数：
use_case（可调用）：异步函数返回UseCaseResultsuccess_handler（可调用，可选）：自定义成功格式化程序error_handler（可调用，可选）：自定义错误格式化程序
退货： 异步可调用，与MCP工具兼容
______________________________________________________________________
bulk_register_tools()
从配置字典中注册多个工具。
参数：
server（FastMCP）：MCP服务器实例tools_config（dict）：工具配置
退货： （tool_name，description）元组列表
______________________________________________________________________
bulk_remove_tools() *（v1.2.0）*
从正在运行的服务器中删除多个工具。
参数：
server（FastMCP）：MCP服务器实例tool_names（list\[str\]）：要删除的工具名称
退货： 词典与 removed, failed，以及 success_rate 钥匙
______________________________________________________________________
bulk_replace_tools() *（v1.2.0）*
原子性地取代了热重新加载的工具。
参数：
server（FastMCP）：MCP服务器实例tools_to_remove（list\[str\]）：要删除的工具tools_to_add（dict）：添加新工具
退货： 带运算结果的词典
______________________________________________________________________
有关API的完整文档，请参阅 API 参考.
______________________________________________________________________
v2.1.1的新增功能
依赖关系更新
- ✅ MCP SDK更新到1.27.0（最新稳定版）
- ✅ 所有开发依赖项已更新到最新版本
- ✅ 构建后端从setuptools切换到孵化器（PEP 621）
- ✅ GitHub Actions工作流现代化（Node.js 24，安装python v5）
以前的亮点
- v2.0.0版本：中断清理--删除死代码、异常、未使用的方法
- v1.3.x：配置管理、错误层次结构、服务器构建器
- v1.2.x：工具生命周期管理（拆卸、更换、检查工具）
- v1.1.x：批量注册，适配器模式基础
看 更改日志.md 查看完整的版本历史记录。
______________________________________________________________________
贡献
欢迎投稿！请看 贡献.md 作为指导方针。
开发设置
# Clone repository
git clone https://github.com/dawsonlp/mcp-commons.git
cd mcp-commons
# Install with dev dependencies
pip install -e ".[dev]"
# Run tests
pytest tests/ -v
# Run linting
black src/ tests/
isort src/ tests/
ruff check src/ tests/______________________________________________________________________
支持
- 📖 文档:
- 🐛 问题:
- 💬 讨论:
______________________________________________________________________
许可证
MIT许可证-请参阅 许可证 了解详情。
______________________________________________________________________
致谢
与 模型上下文协议 通过Anthropic。