"""
Direct MCP Server Implementation
Uses FastMCP framework for stateless HTTP MCP server
"""
import os
import logging
from typing import Any, Dict
logger = logging.getLogger(__name__)
try:
from mcp.server.fastmcp import FastMCP
except ImportError:
logger.warning("FastMCP not available. Install with: pip install mcp")
FastMCP = None
def create_mcp_server(host: str = "0.0.0.0", port: int = 8000):
"""
Create and configure MCP server using FastMCP
Args:
host: Server host
port: Server port
Returns:
Configured FastMCP server instance
"""
if FastMCP is None:
raise ImportError("FastMCP is required. Install with: pip install mcp")
# Create FastMCP server with stateless HTTP (required for AgentCore)
mcp = FastMCP(
host=host,
stateless_http=True # Required for Bedrock AgentCore
)
# Example tool: Add numbers
@mcp.tool()
def add_numbers(a: int, b: int) -> int:
"""Add two numbers together"""
return a + b
# Example tool: Get weather
@mcp.tool()
def get_weather(location: str, units: str = "celsius") -> str:
"""
Get weather information for a location
Args:
location: City name or location
units: Temperature units (celsius or fahrenheit)
"""
# Mock weather data
temp = 22 if units == "celsius" else 72
return f"Weather in {location}: {temp}° {units}, Partly Cloudy"
# Example tool: Lookup data
@mcp.tool()
def lookup_data(query: str, value: str) -> Dict[str, Any]:
"""
Lookup data by query and value
Args:
query: Type of query (e.g., 'user_id', 'order_id')
value: Value to lookup
"""
# Mock data
mock_data = {
'user_id': {
'12345': {'id': '12345', 'name': 'John Doe', 'email': 'john@example.com'}
}
}
if query in mock_data and value in mock_data[query]:
return mock_data[query][value]
return {"error": f"No data found for {query}={value}"}
return mcp
def run_server(host: str = "0.0.0.0", port: int = 8000):
"""
Run the MCP server
Args:
host: Server host
port: Server port
"""
mcp = create_mcp_server(host, port)
# Run the server
# FastMCP handles the server lifecycle
logger.info(f"Starting MCP server on {host}:{port}")
mcp.run(host=host, port=port)
if __name__ == "__main__":
host = os.getenv("MCP_HOST", "0.0.0.0")
port = int(os.getenv("MCP_PORT", "8000"))
run_server(host, port)
