"""Remote MCP Server exposing ADK travel planner tools over HTTP.
This server wraps ADK function tools and exposes them via MCP's
Streamable HTTP transport, suitable for deployment on Cloud Run
or any container platform. Any remote MCP client can connect to it.
Architecture:
Remote MCP Client <--HTTP/SSE--> This Server (Cloud Run) <--wraps--> ADK Tools
Usage:
# Run locally:
python mcp_server_remote.py
# Deploy to Cloud Run:
gcloud run deploy travel-planner-mcp --source . --region us-central1
IMPORTANT: All logging goes to stderr. stdout is not used for protocol
in HTTP mode, but we maintain the practice for consistency.
"""
import json
import logging
import os
import sys
# Ensure the project root is in sys.path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stderr,
)
logger = logging.getLogger("travel-planner-mcp-server")
from dotenv import load_dotenv
from mcp import types as mcp_types
from mcp.server.lowlevel import Server, NotificationOptions
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.mcp_tool.conversion_utils import adk_to_mcp_tool_type
from travel_planner.tools.weather_tools import get_weather
from travel_planner.tools.travel_tools import create_travel_plan
load_dotenv()
# Initialize ADK tools
logger.info("Initializing ADK travel planner tools...")
weather_tool = FunctionTool(get_weather)
travel_plan_tool = FunctionTool(create_travel_plan)
adk_tools = {
weather_tool.name: weather_tool,
travel_plan_tool.name: travel_plan_tool,
}
logger.info("Tools initialized: %s", list(adk_tools.keys()))
# Create MCP Server
app = Server("travel-planner-mcp-server")
@app.list_tools()
async def list_mcp_tools() -> list[mcp_types.Tool]:
"""Advertises the travel planner tools this server exposes."""
logger.info("Received list_tools request.")
tools = []
for adk_tool in adk_tools.values():
mcp_schema = adk_to_mcp_tool_type(adk_tool)
tools.append(mcp_schema)
logger.info("Advertising tool: %s", mcp_schema.name)
return tools
@app.call_tool()
async def call_mcp_tool(
name: str, arguments: dict
) -> list[mcp_types.Content]:
"""Executes tool calls requested by MCP clients."""
logger.info("Received call_tool for '%s' with args: %s", name, arguments)
if name in adk_tools:
try:
adk_tool = adk_tools[name]
result = await adk_tool.run_async(
args=arguments,
tool_context=None,
)
logger.info("Tool '%s' executed successfully.", name)
response_text = json.dumps(result, indent=2)
return [mcp_types.TextContent(type="text", text=response_text)]
except Exception as e:
logger.error("Error executing tool '%s': %s", name, e)
error_text = json.dumps(
{"error": f"Failed to execute tool '{name}': {str(e)}"}
)
return [mcp_types.TextContent(type="text", text=error_text)]
else:
logger.warning("Tool '%s' not found.", name)
error_text = json.dumps(
{"error": f"Tool '{name}' not found. Available: {list(adk_tools.keys())}"}
)
return [mcp_types.TextContent(type="text", text=error_text)]
# Create Streamable HTTP session manager (handles multiple client sessions)
session_manager = StreamableHTTPSessionManager(
app=app,
json_response=False,
stateless=True,
)
# Build Starlette ASGI app with CORS and mount MCP at /mcp
starlette_app = Starlette(
debug=False,
middleware=[
Middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
allow_headers=["*"],
expose_headers=["mcp-session-id"],
),
],
routes=[
Mount("/mcp", app=session_manager.handle_request),
],
)
if __name__ == "__main__":
import uvicorn
port = int(os.environ.get("PORT", 8080))
logger.info("Starting Travel Planner MCP Server on port %d", port)
logger.info("MCP endpoint: http://0.0.0.0:%d/mcp", port)
logger.info("Exposing tools: %s", list(adk_tools.keys()))
uvicorn.run(
starlette_app,
host="0.0.0.0",
port=port,
log_level="info",
)
