MCP共享

一个Python库，为构建具有较少样板和一致模式的模型上下文协议（MCP）服务器提供可重用的基础设施。

](https://pypi.org/project/mcp-commons/) ](https://pypi.org/project/mcp-commons/) ![License: MIT](https://opensource.org/licenses/MIT)

概述

MCP Commons为构建可维护的MCP服务器提供了架构模式：

主要价值（90%）:

适配器模式 -将业务逻辑与MCP协议解耦，以实现多接口重用
用例结果 -所有操作中一致的错误处理模式

便利功能（10%）:

批量操作 -带有错误报告的配置驱动工具注册
工具生命周期 -通过FastMCP的add_tool（）和remove_tool（

基于FastMCP构建：mcp-commons是FastMCP现有方法的精简封装。它不会取代FastMCP的功能——它提供了架构模式和便利的包装器，使您的代码更易于维护。

当前版本: 2.1.1 | 最新动态 | 更新日志

______________________________________________________________________

为什么存在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在这个基本的设计决策上犯了错误。 适配器模式并不“好用”，它对于任何非平凡应用程序中的正确架构都是必不可少的。

______________________________________________________________________

安装

需求

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 （可调用）：异步函数返回 UseCaseResult
success_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。

MCP Commons

快速接入

详细介绍

MCP共享

概述

为什么存在mcp公共资源

装饰师的问题

解决方案：适配器模式

建筑效益

目录

安装

需求

从PyPI安装

安装用于开发

FastMCP提供什么与mcp commons添加什么

快速开始

1.基本适配器模式

2.批量注册

3.工具管理（v1.2.0）

核心功能

工具适配器

基本用法

错误处理

批量注册

配置字典

元组格式（简单）

使用适配器模式

工具管理（v1.2.0）

删除工具

更换工具（热重新加载）

有条件移除

工具检查

高级用法

自定义错误处理程序

验证和记录

测试您的工具

API 参考

核心功能

create_mcp_adapter()

bulk_register_tools()

bulk_remove_tools() *（v1.2.0）*

bulk_replace_tools() *（v1.2.0）*

v2.1.1的新增功能

依赖关系更新

以前的亮点

贡献

开发设置

支持

许可证

致谢

目录标签

接入字段

权限和风险

来源信息

继续浏览同类 MCP

`create_mcp_adapter()`

`bulk_register_tools()`

`bulk_remove_tools()` （v1.2.0）

`bulk_replace_tools()` （v1.2.0）