MCP Commons
A Python library providing reusable infrastructure for building Model Context Protocol (MCP) servers with less boilerplate and consistent patterns.
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:
- DRY Principle - One business function, multiple interfaces
- Separation of Concerns - Business logic independent of transport
- Framework Independence - No coupling to MCP SDK, FastAPI, Click, etc.
- Easy Testing - Test pure functions without framework context
- 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 returningUseCaseResultcustom_success_handler(callable, optional, keyword-only): Receives the result'sdata; returns the response dictcustom_error_handler(callable, optional, keyword-only): Receives a raisedException; 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 instancetools_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 instancetool_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 instancetools_to_remove(list[str]): Tools to removetools_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 oldsuccessclassmethod collided with thesuccessfield, leaving the field without a real default; the rename restores correct dataclass semantics. The.success/.data/.error/.detailsattributes are unchanged. - ⚠️
run_mcp_server()andcreate_mcp_app()no longer accept the unusedconfigparameter (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-onlysetup_logging) and thesetup_loggingstderr 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
- 📖 Documentation: GitHub Wiki
- 🐛 Issues: GitHub Issues
- 💬 Discussions: GitHub Discussions
License
MIT License - see LICENSE for details.
Acknowledgments
Built with the Model Context Protocol by Anthropic.
Metadata
Release files for mcp-commons 3.0.1
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| mcp_commons-3.0.1.tar.gz | 43.1 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| mcp_commons-3.0.1-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 68.0 kB
Release files / mcp_commons-3.0.1.tar.gz
| Download URL | mcp_commons-3.0.1.tar.gz |
|---|---|
| Size | 43.1 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
3bf5c041c03eb8d39dde9c7fd6b3a251c4366fe34e9545660d16d9538c238c04
|
|
BLAKE2b-256 checksum How to use checksums |
b7131eaf05ee7440d7ebb2563823694655710bf2732f74ebcdf28f0488ed8214
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.1.0 CPython/3.13.13
|
Release files / mcp_commons-3.0.1-py3-none-any.whl
| Download URL | mcp_commons-3.0.1-py3-none-any.whl |
|---|---|
| Size | 24.9 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
2b89609aa615c80a1f546ad31e981cb8b05800f971de9cf40c618a7f5f519c4e
|
|
BLAKE2b-256 checksum How to use checksums |
2bb04236ad4bceb632925bf40426090b7606caa0d9a1962518474078f81b44cb
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.1.0 CPython/3.13.13
|