This document explains the advanced architectural pattern used in this project where REST API and MCP server share the same metadata with zero duplication.
Traditional approaches to supporting multiple interfaces lead to duplication:
# REST endpoint
@app.route("/api/tasks", methods=["POST"])
async def create_task():
"""Create a new task""" # Description #1
data = await request.get_json()
if "title" not in data: # Validation #1
return {"error": "Title required"}, 400
# ... implementation
# MCP tool
@mcp.tool()
async def create_task_tool():
"""Create a new task""" # Description #2 (duplicate!)
return Tool(
name="create_task",
description="Create a new task", # Description #3 (duplicate!)
inputSchema={ # Schema (duplicate validation!)
"type": "object",
"properties": {
"title": {"type": "string", "description": "Task title"}
},
"required": ["title"]
}
)
# ... different implementation using same business logicProblems:
- Descriptions defined 3 times
- Validation defined 2 times
- Schema defined 2 times
- Easy to get out of sync
- More code to maintain
from api_decorators import operation, Parameter
@operation(
name="create_task",
description="Create a new task", # DEFINED ONCE
parameters=[
Parameter("title", "string", "Task title", required=True) # DEFINED ONCE
],
http_method="POST",
http_path="/api/tasks"
)
async def op_create_task(title: str):
return tasks_service.create_task(title)Benefits:
- ✅ Description defined once
- ✅ Schema defined once
- ✅ Automatically generates both REST and MCP
- ✅ Cannot get out of sync
- ✅ Less code to maintain
1. The Decorator System (api_decorators.py)
The @operation decorator captures all metadata:
@operation(
name="create_task", # Used for MCP tool name
description="...", # Used for both REST docs and MCP description
parameters=[...], # Converted to JSON schema for MCP
http_method="POST", # Used for REST routing
http_path="/api/tasks" # Used for REST routing
)
async def op_create_task(title: str, description: str = ""):
return tasks_service.create_task(title, description)This decorator:
- Stores the operation in a global registry
- Captures metadata that works for both REST and MCP
- Returns the original function unchanged
2. REST Route Generation (app.py)
REST routes call the decorated functions:
@app.route("/api/tasks", methods=["POST"])
async def rest_create_task():
"""REST wrapper for create_task operation."""
data = await request.get_json()
# Validate using operation metadata
if not data or "title" not in data:
return jsonify({"error": "Title is required"}), 400
# Call the operation handler
task, error = await op_create_task(
title=data["title"],
description=data.get("description", "")
)
if error:
return jsonify({"error": error}), 400
return jsonify(task), 2013. MCP Tool Generation (app.py)
MCP tools are auto-generated from the same decorators:
@app.route("/mcp", methods=["POST"])
async def mcp_json_rpc():
method = data.get("method")
if method == "tools/list":
# Auto-generate tool list from @operation decorators
tools = get_mcp_tools() # Reads from decorator registry
return jsonify({"jsonrpc": "2.0", "result": {"tools": tools}})
elif method == "tools/call":
tool_name = params.get("name")
arguments = params.get("arguments", {})
# Get the operation handler
op = get_operation(tool_name)
# Call the same handler that REST uses!
result = await op.handler(**arguments)
return jsonify({"jsonrpc": "2.0", "result": {...}})4. Shared Business Logic (tasks/service.py)
Both interfaces call the same business logic:
# In tasks/service.py
def create_task(title: str, description: str = "") -> tuple[dict | None, str | None]:
"""Core business logic - used by all interfaces."""
# Validation
is_valid, error = validate_task_data({"title": title})
if not is_valid:
return None, error
# Create task
task = create_task_data(title, description)
# Save
saved_task = save_task(task)
return saved_task, None┌─────────────────────────────────────────────────────────┐
│ Interfaces Layer │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ REST Routes │ │ MCP JSON-RPC │ │
│ │ (HTTP Wrappers) │ │ (HTTP/POST) │ │
│ └────────┬────────┘ └────────┬────────┘ │
└───────────┼──────────────────────────┼──────────────────┘
│ │
└──────────┬───────────────┘
│
┌─────────────▼─────────────┐
│ @operation Decorators │
│ (api_decorators.py) │
│ • Metadata registry │
│ • Schema generation │
│ • Operation routing │
└─────────────┬──────────────┘
│
┌─────────────▼─────────────┐
│ Operation Handlers │
│ (op_create_task, etc.) │
│ • Parameter handling │
│ • Call business logic │
└─────────────┬──────────────┘
│
┌─────────────▼─────────────┐
│ Business Logic Layer │
│ (tasks/service.py) │
│ • Calculations │
│ • Actions │
│ • Data │
└────────────────────────────┘
Metadata is defined exactly once. Change it once, both interfaces update.
REST and MCP cannot have different descriptions or schemas - they're generated from the same source.
No need for inter-process communication. Both interfaces share the same memory and business logic.
Want to add GraphQL? WebSocket? CLI? Just add another interface layer that uses the same operations.
TypeScript-like type hints in Python decorators help catch errors early.
To add a new operation that's exposed as both REST and MCP:
@operation(
name="mark_all_complete",
description="Mark all tasks as completed",
parameters=[],
http_method="POST",
http_path="/api/tasks/complete-all"
)
async def op_mark_all_complete() -> dict:
"""Mark all tasks as completed."""
count = tasks_service.mark_all_complete()
return {"marked_complete": count}
# Add REST wrapper
@app.route("/api/tasks/complete-all", methods=["POST"])
async def rest_mark_all_complete():
result = await op_mark_all_complete()
return jsonify(result)
# MCP automatically picks it up from the decorator!
# No additional MCP code needed!curl -X POST http://localhost:5001/api/tasks \
-H 'Content-Type: application/json' \
-d '{"title":"REST Task","description":"Created via REST"}'curl -X POST http://localhost:5001/mcp \
-H 'Content-Type: application/json' \
-d '{"jsonrpc":"2.0","method":"tools/call","params":{"name":"create_task","arguments":{"title":"MCP Task","description":"Created via MCP"}},"id":1}'curl http://localhost:5001/api/tasks
# Both "REST Task" and "MCP Task" appear!- Deep Modules:
tasks.servicehas simple interface, complex implementation - Information Hiding: Business logic hidden behind operation decorators
- Define Errors Out of Existence: Cannot have inconsistent metadata between interfaces
- Separate Calculations from Actions: Pure functions vs I/O in
tasks.service - Stratified Design: Clear layers (interface → operations → business logic)
- Minimize Side Effects: Operations are predictable and testable
| Approach | Lines of Code | Duplication | Consistency | Maintenance |
|---|---|---|---|---|
| Separate REST + MCP | ~500 | High | Manual sync | Hard |
| Shared Business Logic | ~400 | Medium | Better | Medium |
| Unified Operations (Ours) | ~350 | None | Guaranteed | Easy |
This pattern makes it trivial to add:
- GraphQL: Read operations, call handlers
- gRPC: Define protobuf, call handlers
- WebSocket: Real-time events, call handlers
- CLI: Argparse interface, call handlers
- Message Queue: Consume messages, call handlers
All without duplicating business logic or metadata!
- See app.py for the implementation
- See api_decorators.py for the decorator system
- See tasks/service.py for business logic
- See README.md for usage instructions