Skip to main content

MCP Commons

A Python library providing reusable infrastructure for building Model Context Protocol (MCP) servers with less boilerplate and consistent patterns.

PyPI version Python versions License: MIT

Overview

MCP Commons provides architectural patterns for building maintainable MCP servers:

Primary Value (90%):

  • Adapter Pattern - Decouple business logic from MCP protocol for multi-interface reuse
  • UseCaseResult - Consistent error handling pattern across all operations

Convenience Features (10%):

  • Bulk Operations - Config-driven tool registration with error reporting
  • Tool Lifecycle - Batch operations over FastMCP's add_tool() and remove_tool()

Built on FastMCP: mcp-commons is a thin wrapper over FastMCP's existing methods. It doesn't replace FastMCP's capabilities - it provides architectural patterns and convenience wrappers to make your code more maintainable.

Current Version: 3.0.1 | What's New | Changelog


Why mcp-commons Exists

The Problem with Decorators

The MCP SDK uses decorators (@server.tool()) to register functions as tools. While the goal of making function exposure easy was admirable, decorators were the wrong mechanism for this purpose.

Decorators should add cross-cutting concerns (caching, authentication, logging) that apply regardless of how a function is used. They should not specify usage contexts (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)

The Solution: Adapter Pattern

Adapter/wrapper functions provide the same ease of use while maintaining proper separation of concerns. Your business logic stays pure and framework-agnostic, while thin adapters handle protocol translation.

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

Architectural Benefits

This adapter pattern enables:

  1. DRY Principle - One business function, multiple interfaces
  2. Separation of Concerns - Business logic independent of transport
  3. Framework Independence - No coupling to MCP SDK, FastAPI, Click, etc.
  4. Easy Testing - Test pure functions without framework context
  5. Future-Proof - When MCP SDK v2.0 changes, only adapters need updates

mcp-commons exists because the MCP SDK got this fundamental design decision wrong. The adapter pattern isn't "nice to have" - it's essential for proper architecture in any non-trivial application.


Table of Contents


Installation

Requirements

  • Python: 3.11+ (3.13 recommended)
  • MCP SDK: 1.28.1 or newer within the 1.x release line (>=1.28.1,<2.0.0)
  • Dependencies: Pydantic 2.13.4+, PyYAML 6.0.3+

Install from PyPI

pip install mcp-commons

Install for Development

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

What FastMCP Provides vs What mcp-commons Adds

Feature FastMCP (SDK) mcp-commons
Tool registration server.add_tool(func, name, desc) Config-driven bulk wrapper
Tool removal server.remove_tool(name) (v1.17.0+) Batch wrapper with reporting
Decorators @server.tool() decorator ❌ We don't use decorators
Adapter pattern ❌ Not provided ✅ Core feature - decouples logic
UseCaseResult ❌ Not provided ✅ Consistent error handling
Error reporting Exceptions on failure Success/failure batch reports
Tool managers ToolManager, ResourceManager, etc. ✅ We use FastMCP's managers

Key Point: mcp-commons doesn't replace FastMCP - it builds on it. We use FastMCP's add_tool() and remove_tool() methods internally, adding convenience wrappers and architectural patterns on top.


Quick Start

1. Basic Adapter Pattern

Convert your async functions to MCP tools:

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)

2. Bulk Registration

Register multiple tools at once:

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. Tool Management (v1.2.0)

Dynamically manage tools at runtime:

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

Core Features

Tool Adapters

The adapter pattern automatically handles the conversion between your business logic and MCP tool format.

Basic Usage

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)

Error Handling

UseCaseResult gives every operation a consistent success/failure shape (the adapter then unwraps a successful result's data and reports failures as a {"success": False, ...} dict):

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

Bulk Registration

Convenience wrappers over FastMCP's add_tool() method for config-driven registration:

What it actually does:

# 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

Why use it: Config-driven API + batch error handling instead of manual loops.

Configuration Dictionary

tools_config = {
    "tool_name": {
        "function": async_function,
        "description": "Tool description",
        # Optional metadata
    }
}

registered = bulk_register_tools(server, tools_config)

Tuple Format (Simple)

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)

With Adapter Pattern

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
)

Tool Management (v1.2.0)

New in version 1.2.0: Convenience wrappers for batch tool operations.

What it actually does: Loops over FastMCP's remove_tool() method (added in SDK v1.17.0) with error reporting:

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

Why use it: Batch operations + detailed success/failure reporting instead of manual loops.

Remove Tools

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

Replace Tools (Hot Reload)

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

Conditional Removal

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

Tool Inspection

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

Advanced Usage

Custom Error Handlers

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
)

Validation and Logging

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

Testing Your Tools

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

API Reference

Core Functions

create_mcp_adapter()

Converts an async function to an MCP-compatible tool adapter.

Parameters:

  • use_case_method (callable): Async function returning UseCaseResult
  • custom_success_handler (callable, optional, keyword-only): Receives the result's data; returns the response dict
  • custom_error_handler (callable, optional, keyword-only): Receives a raised Exception; returns the response dict

Returns: Async callable compatible with MCP tools


bulk_register_tools()

Registers multiple tools from a configuration dictionary.

Parameters:

  • server (FastMCP): MCP server instance
  • tools_config (dict): Tool configurations

Returns: List of (tool_name, description) tuples


bulk_remove_tools() (v1.2.0)

Removes multiple tools from a running server.

Parameters:

  • server (FastMCP): MCP server instance
  • tool_names (list[str]): Names of tools to remove

Returns: Dictionary with removed, failed, and success_rate keys


bulk_replace_tools() (v1.2.0)

Atomically replaces tools for hot-reloading.

Parameters:

  • server (FastMCP): MCP server instance
  • tools_to_remove (list[str]): Tools to remove
  • tools_to_add (dict): New tools to add

Returns: Dictionary with operation results


For complete API documentation, see API Reference.


What's New in v3.0.1

Compatibility fix

  • ✅ MCP SDK constrained to the compatible 1.x release line (>=1.28.1,<2.0.0).

Previous v3.0.0 highlights

  • ⚠️ UseCaseResult.success() / .failure() are renamed to .succeeded() / .failed(). The old success classmethod collided with the success field, leaving the field without a real default; the rename restores correct dataclass semantics. The .success / .data / .error / .details attributes are unchanged.
  • ⚠️ run_mcp_server() and create_mcp_app() no longer accept the unused config parameter (it was silently ignored).
  • ⚠️ run_mcp_server() no longer configures logging itself — that is the caller's job. run_cli() still sets up logging before delegating, so the common entry point is unaffected.
  • ✅ The convenience adapters are now exported from the top-level package: create_query_adapter, create_command_adapter, create_validation_adapter, create_mock_adapter, get_adapter_stats.
  • ⬆️ MCP SDK 1.28.1, pytest 9.1.1, pytest-asyncio 1.4.0, ruff 0.15.21.

Previous Highlights

  • v2.2.x: Server helpers (find_server_config, run_cli, create_config(server_name=...), keyword-only setup_logging) and the setup_logging stderr default fix
  • v2.1.x: Dependency refresh (MCP SDK 1.27.1, pydantic 2.13.4), hatchling build backend
  • v2.0.0: Breaking cleanup -- removed dead code, exceptions, unused methods
  • v1.3.x: Configuration management, error hierarchy, server builder
  • v1.2.x: Tool lifecycle management (remove, replace, inspect tools)
  • v1.1.x: Bulk registration, adapter pattern foundations

See CHANGELOG.md for complete version history.


Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

Development Setup

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

Support


License

MIT License - see LICENSE for details.


Acknowledgments

Built with the Model Context Protocol by Anthropic.

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

mcp_commons-3.0.1.tar.gz (43.1 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

mcp_commons-3.0.1-py3-none-any.whl (24.9 kB view details)

Uploaded Python 3

File details

Details for the file mcp_commons-3.0.1.tar.gz.

File metadata

  • Download URL: mcp_commons-3.0.1.tar.gz
  • Upload date:
  • Size: 43.1 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.13.13

File hashes

Hashes for mcp_commons-3.0.1.tar.gz
Algorithm Hash digest
SHA256 3bf5c041c03eb8d39dde9c7fd6b3a251c4366fe34e9545660d16d9538c238c04
MD5 6f5df7faac21a36bfc8a408f4711638a
BLAKE2b-256 b7131eaf05ee7440d7ebb2563823694655710bf2732f74ebcdf28f0488ed8214

See more details on using hashes here.

File details

Details for the file mcp_commons-3.0.1-py3-none-any.whl.

File metadata

  • Download URL: mcp_commons-3.0.1-py3-none-any.whl
  • Upload date:
  • Size: 24.9 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.13.13

File hashes

Hashes for mcp_commons-3.0.1-py3-none-any.whl
Algorithm Hash digest
SHA256 2b89609aa615c80a1f546ad31e981cb8b05800f971de9cf40c618a7f5f519c4e
MD5 ea9b571c684d9b3a59fc23fd7c505b85
BLAKE2b-256 2bb04236ad4bceb632925bf40426090b7606caa0d9a1962518474078f81b44cb

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page