MCP服务器构建工具库

# ❌ 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

pip install mcp-commons

git clone https://github.com/dawsonlp/mcp-commons.git
cd mcp-commons
pip install -e ".[dev]"

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.succeeded({
        "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)

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")

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"
        }
    }
)

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.succeeded(metrics)
    except DatasetNotFoundError as e:
        return UseCaseResult.failed(f"Dataset not found: {e}")
    except Exception as e:
        return UseCaseResult.failed(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.succeeded({"status": "completed", "value": 42})
# -> UseCaseResult(success=True, data={"status": "completed", "value": 42}, error=None)

# Failure response  
UseCaseResult.failed("Invalid input parameters")
# -> UseCaseResult(success=False, data=None, error="Invalid input parameters")

# 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

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

# Each tuple is (function, name, description)
tools = [
    (list_items_function, "list_items", "List all items"),
    (get_item_function, "get_item", "Get an item by ID"),
    (create_item_function, "create_item", "Create a new item"),
]

bulk_register_tuple_format(server, tools)

from mcp_commons import bulk_register_with_adapter_pattern

# Each entry maps a tool name to {"use_case": <async callable returning
# UseCaseResult>, "description": <str>}.
use_cases = {
    "validate_data": {"use_case": validate_data_use_case, "description": "Validate a dataset"},
    "process_data": {"use_case": process_data_use_case, "description": "Process a dataset"},
    "export_data": {"use_case": export_data_use_case, "description": "Export a dataset"},
}

bulk_register_with_adapter_pattern(
    server,
    use_cases,
    adapter_function=create_mcp_adapter
)

# 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(data):
    """Receives the successful result's `.data` value."""
    return {
        "status": "success",
        "payload": data,
        "timestamp": datetime.now().isoformat()
    }

def custom_error_handler(error):
    """Receives an Exception raised by the use case."""
    return {
        "status": "error",
        "message": str(error),
        "timestamp": datetime.now().isoformat()
    }

adapted = create_mcp_adapter(
    my_function,
    custom_success_handler=custom_success_handler,
    custom_error_handler=custom_error_handler
)

from mcp_commons import validate_tools_config, log_registration_summary

# Validate before registering (returns a report dict; it never raises)
validation = validate_tools_config(tools_config)
if not validation["valid"]:
    print(f"Invalid configuration: {validation['issues']}")
    
# 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.succeeded({"results": ["doc1", "doc2"]})
    
    adapted = create_mcp_adapter(mock_search)
    result = await adapted(query="test")
    
    # On success the adapter returns the result's data directly
    assert result == {"results": ["doc1", "doc2"]}
    assert len(result["results"]) == 2

# 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/