MCPAPIRouter
MCPAPIRouter provides a FastAPI-like decorator API for registering MCP tools, prompts, and resources. It handles JSON-RPC protocol communication and automatic schema generation.
Class Definition
from agentor.mcp.api_router import MCPAPIRouter
class MCPAPIRouter
Constructor
router = MCPAPIRouter(
prefix="/mcp",
name="agentor-mcp-server",
version="0.1.0",
instructions=None,
website_url=None,
icons=None,
dependencies=None,
)
URL prefix for MCP endpoints
name
str
default:"agentor-mcp-server"
Server name returned in MCP initialize response
Instructions for using the server, sent to clients during initialization
Server icons (from mcp.types.Icon)
FastAPI dependencies to apply to all MCP endpoints
Decorators
Register a tool that can be called by MCP clients.
@router.tool(
name=None,
description=None,
input_schema=None,
)
def tool_function(param1: str, param2: int) -> str:
"""Tool description from docstring"""
return "result"
Tool name. Defaults to function name.
Tool description shown to LLMs. Defaults to function docstring.
JSON schema for tool parameters. Auto-generated from function signature if not provided.
@app.tool(description="Get weather for a location")
def get_weather(location: str) -> str:
"""Get current weather for a location"""
return f"Weather in {location}: Sunny, 72°F"
from fastapi import Depends
from agentor.mcp.api_router import Context, get_context
@app.tool()
def authenticated_tool(
location: str,
ctx: Context = Depends(get_context)
) -> str:
"""Tool that accesses request context"""
user_agent = ctx.headers.get("user-agent")
session_id = ctx.cookies.get("session_id")
return f"Processing {location} for session {session_id}"
@app.tool()
async def fetch_data(url: str) -> str:
"""Asynchronously fetch data from URL"""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
@prompt()
Register a prompt template that can be retrieved by MCP clients.
@router.prompt(
name=None,
description=None,
arguments=None,
)
def prompt_function(param1: str, param2: str = "default") -> str:
"""Prompt description"""
return "prompt text"
Prompt name. Defaults to function name.
Prompt description. Defaults to function docstring.
List of argument definitions. Auto-generated from function signature if not provided.Each argument dict should have:
name (str): Argument namedescription (str): Argument descriptionrequired (bool): Whether the argument is required
Example: Basic Prompt
@app.prompt(description="Generate a greeting")
def greeting(name: str, style: str = "formal") -> str:
"""Generate a personalized greeting"""
if style == "formal":
return f"Good day, {name}. How may I assist you today?"
else:
return f"Hey {name}! What's up?"
Example: Prompt with Message Structure
@app.prompt()
def code_review(code: str, language: str) -> list:
"""Generate a code review prompt"""
return [
{
"role": "user",
"content": {
"type": "text",
"text": f"Review this {language} code:\n\n{code}"
}
}
]
@resource()
Register a resource that can be read by MCP clients.
@router.resource(
uri="resource://path",
name=None,
description=None,
mime_type=None,
)
def resource_function(uri: str) -> str:
"""Resource description"""
return "resource content"
Resource URI identifier. Used by clients to request the resource.
Resource display name. Defaults to URI.
Resource description. Defaults to function docstring.
MIME type of resource content (e.g., “application/json”, “text/plain”)
Example: Basic Resource
@app.resource(
uri="config://settings",
name="Settings",
mime_type="application/json"
)
def get_settings(uri: str) -> str:
"""Get application settings"""
return '{"theme": "dark", "language": "en"}'
Example: Dynamic Resource
@app.resource(
uri="file://data",
name="Data File",
mime_type="text/plain"
)
def get_file_data(uri: str) -> dict:
"""Get file data with metadata"""
return {
"uri": uri,
"mimeType": "text/plain",
"text": "File content here"
}
@method()
Register a custom MCP JSON-RPC method handler.
@router.method("custom/method")
async def custom_handler(body: dict):
"""Handle custom MCP method"""
params = body.get("params", {})
return {"result": "custom response"}
JSON-RPC method name to handle
Context and Dependencies
Context
Access request-level data in tool functions.
from agentor.mcp.api_router import Context, get_context
from fastapi import Depends
@app.tool()
def my_tool(location: str, ctx: Context = Depends(get_context)) -> str:
user_agent = ctx.headers.get("user-agent")
session_id = ctx.cookies.get("session_id")
return f"Processing {location}"
HTTP headers from the request
get_context()
Dependency function to retrieve the current request context.
def get_context() -> Context
Context object with headers and cookies. Returns empty context if called outside a request.
Get HTTP headers from the current request.
def get_headers() -> Dict[str, str]
get_cookies()
Get cookies from the current request.
def get_cookies() -> Dict[str, str]
get_token()
Extract bearer token from Authorization header.
Returns the token without the “Bearer ” prefix, or None if no token is found.
Methods
get_fastapi_router()
Get the underlying FastAPI router instance.
fastapi_router = router.get_fastapi_router()
APIRouter instance that can be included in FastAPI applications.
Complete Example
from agentor.mcp.api_router import MCPAPIRouter, Context, get_context
from fastapi import Depends, FastAPI
# Create router
router = MCPAPIRouter(
prefix="/mcp",
name="example-server",
version="1.0.0",
instructions="Example MCP server with tools, prompts, and resources",
)
# Register tool
@router.tool(description="Calculate sum of two numbers")
def add(a: int, b: int) -> int:
"""Add two numbers together"""
return a + b
# Register tool with context
@router.tool()
def user_info(ctx: Context = Depends(get_context)) -> str:
"""Get user agent from request"""
return ctx.headers.get("user-agent", "unknown")
# Register prompt
@router.prompt(description="Code review template")
def review_prompt(language: str, code: str) -> str:
"""Generate code review prompt"""
return f"Review this {language} code:\n\n{code}"
# Register resource
@router.resource(
uri="docs://readme",
name="README",
mime_type="text/markdown"
)
def readme(uri: str) -> str:
"""Get README content"""
return "# Example Server\n\nThis is an example MCP server."
# Use with FastAPI
app = FastAPI()
app.include_router(router.get_fastapi_router())
See Also