"""
Example tool demonstrating the pattern for creating MCP tools.
This is a simple example tool that shows:
- How to use the register pattern
- How to use the envelope for responses
- How to add proper type hints and documentation
- How to handle errors
You can copy this as a template for new tools.
"""
from __future__ import annotations
from typing import Optional
from uuid import uuid4
from mcp.server.fastmcp import FastMCP
from src.core.tool_registry import ContextAwareToolRegistry
from src.core.tool_context import PageType, EntityType, ToolContextPolicy
from src.core.domain_policies import SYSTEM_POLICIES
from pydantic import BaseModel, Field
from src.core import wrap_response
from src.observability import get_logger
logger = get_logger(__name__)
# Input schema (optional but recommended for validation)
class ExampleInput(BaseModel):
"""Input parameters for the example tool."""
message: str = Field(description="Message to echo back")
uppercase: bool = Field(default=False, description="Convert to uppercase")
trace_id: Optional[str] = Field(default=None, description="Trace ID for request tracking")
def register(registry: ContextAwareToolRegistry) -> None:
"""
Register example tools with the MCP server.
This function is called automatically by the auto-discovery system.
"""
@registry.register_tool(
name="echo",
policy=SYSTEM_POLICIES["echo"],
description="Echo back a message (example tool for testing)"
)
async def echo(
message: str,
uppercase: bool = False,
trace_id: Optional[str] = None,
) -> dict:
"""
Echo back a message (example tool for testing).
This is a simple example tool that demonstrates the proper pattern
for creating MCP tools with validation, logging, and error handling.
Args:
message: The message to echo back
uppercase: Whether to convert to uppercase
trace_id: Optional trace ID for request tracking
Returns:
Standard envelope with the echoed message
"""
t_id = trace_id or str(uuid4())
logger.info(
"echo_tool_called",
trace_id=t_id,
message_length=len(message),
uppercase=uppercase
)
try:
# Process the message
result = message.upper() if uppercase else message
# Return using standard envelope
return wrap_response(
tool_name="echo",
trace_id=t_id,
data={
"original": message,
"result": result,
"was_uppercased": uppercase,
}
)
except Exception as e:
logger.error(
"echo_tool_failed",
trace_id=t_id,
error=str(e),
error_type=type(e).__name__
)
# Return error in envelope
return wrap_response(
tool_name="echo",
trace_id=t_id,
data={},
missing_reason=f"Tool execution failed: {str(e)}"
)
@registry.register_tool(
name="get_system_info",
policy=SYSTEM_POLICIES["get_system_info"],
description="Get system information (public tool for health checks)"
)
async def get_system_info(trace_id: Optional[str] = None) -> dict:
"""
Get system information (public tool for health checks).
This tool provides basic system information and can be used
to verify the MCP server is running correctly.
"""
t_id = trace_id or str(uuid4())
from src.config import settings
return wrap_response(
tool_name="get_system_info",
trace_id=t_id,
data={
"app": "Updation MCP Server",
"version": "1.0.0",
"environment": settings.environment,
"llm_provider": settings.llm_provider,
"status": "operational",
}
)
