"""MCP Server exposing ADK travel planner tools.
This server wraps ADK function tools (get_weather, create_travel_plan) and
exposes them via the Model Context Protocol (MCP), making them accessible
to any MCP-compliant client application.
Architecture:
MCP Client <--MCP Protocol--> This Server <--wraps--> ADK FunctionTools
Usage:
# Run directly (stdio transport):
python mcp_server.py
# Or use with an ADK agent as MCP client (see adk_client_agent/)
IMPORTANT: MCP stdio transport uses stdout for JSON-RPC messages ONLY.
All logging MUST go to stderr. Never use print() in this file.
"""
import asyncio
import json
import logging
import os
import sys
# Ensure the project root is in sys.path so `travel_planner` package is importable
# even when this script is launched as a subprocess from a different working directory.
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
# Configure logging to stderr (stdout is reserved for MCP protocol)
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.models import InitializationOptions
import mcp.server.stdio
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 to expose via MCP
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 request 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)]
async def run_mcp_stdio_server():
"""Runs the MCP server over standard input/output transport."""
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
logger.info("Starting MCP handshake with client...")
await app.run(
read_stream,
write_stream,
InitializationOptions(
server_name=app.name,
server_version="0.1.0",
capabilities=app.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
logger.info("MCP connection closed.")
def main():
"""Entry point for the MCP server."""
logger.info("Launching Travel Planner MCP Server (stdio transport)...")
logger.info("Exposing tools: %s", list(adk_tools.keys()))
try:
asyncio.run(run_mcp_stdio_server())
except KeyboardInterrupt:
logger.info("MCP Server stopped by user.")
except Exception as e:
logger.error("MCP Server encountered an error: %s", e)
finally:
logger.info("MCP Server process exiting.")
if __name__ == "__main__":
main()
