"""
MCP Protocol Handler
Handles translation between MCP protocol and backend services (Lambda/REST)
"""
import json
import logging
from typing import Any, Dict, List, Optional
from enum import Enum
logger = logging.getLogger(__name__)
class ToolType(str, Enum):
"""Type of tool backend"""
LAMBDA = "lambda"
REST = "rest"
class MCPProtocolHandler:
"""Handles MCP protocol translation and message processing"""
def __init__(self):
self.session_id: Optional[str] = None
def parse_mcp_request(self, request_body: Dict[str, Any]) -> Dict[str, Any]:
"""
Parse incoming MCP request
Args:
request_body: Raw request body
Returns:
Parsed request with method and parameters
"""
try:
method = request_body.get("method", "")
params = request_body.get("params", {})
id = request_body.get("id")
return {
"method": method,
"params": params,
"id": id,
"jsonrpc": request_body.get("jsonrpc", "2.0")
}
except Exception as e:
logger.error(f"Error parsing MCP request: {e}")
raise ValueError(f"Invalid MCP request format: {e}")
def create_mcp_response(self, request_id: Any, result: Any = None, error: Optional[Dict] = None) -> Dict[str, Any]:
"""
Create MCP-compliant response
Args:
request_id: Request ID from original request
result: Response result (if success)
error: Error information (if failure)
Returns:
MCP response dictionary
"""
response = {
"jsonrpc": "2.0",
"id": request_id
}
if error:
response["error"] = error
else:
response["result"] = result
return response
def create_tool_list_response(self, tools: List[Dict[str, Any]], request_id: Any) -> Dict[str, Any]:
"""
Create response for tools/list method
Args:
tools: List of available tools
request_id: Request ID
Returns:
MCP response with tools list
"""
return self.create_mcp_response(
request_id=request_id,
result={
"tools": tools
}
)
def create_tool_call_response(self, result: Any, request_id: Any) -> Dict[str, Any]:
"""
Create response for tools/call method
Args:
result: Tool execution result
request_id: Request ID
Returns:
MCP response with tool result
"""
return self.create_mcp_response(
request_id=request_id,
result={
"content": [
{
"type": "text",
"text": json.dumps(result) if not isinstance(result, str) else result
}
]
}
)
def create_error_response(self, code: int, message: str, request_id: Any, data: Optional[Any] = None) -> Dict[str, Any]:
"""
Create error response
Args:
code: Error code
message: Error message
request_id: Request ID
data: Additional error data
Returns:
MCP error response
"""
error = {
"code": code,
"message": message
}
if data:
error["data"] = data
return self.create_mcp_response(
request_id=request_id,
error=error
)
def extract_tool_arguments(self, params: Dict[str, Any]) -> tuple[str, Dict[str, Any]]:
"""
Extract tool name and arguments from MCP params
Args:
params: MCP request parameters
Returns:
Tuple of (tool_name, arguments)
"""
name = params.get("name", "")
arguments = params.get("arguments", {})
return name, arguments
def format_tool_definition(self, name: str, description: str, input_schema: Dict[str, Any]) -> Dict[str, Any]:
"""
Format tool definition for MCP tools/list response
Args:
name: Tool name
description: Tool description
input_schema: JSON schema for tool inputs
Returns:
Formatted tool definition
"""
return {
"name": name,
"description": description,
"inputSchema": input_schema
}
