"""MCP Server setup and configuration."""
import logging
from collections.abc import Awaitable, Callable
from typing import Any
from mcp.server import Server
from mcp.types import TextContent, Tool
from jana_mcp.client import JanaClient
from jana_mcp.config import Settings
from jana_mcp.tools import (
ALL_TOOLS,
execute_air_quality,
execute_data_summary,
execute_emissions,
execute_nearby,
execute_system_health,
execute_trends,
)
from jana_mcp.tools.response import create_error_response, serialize_response
logger = logging.getLogger(__name__)
# Tool dispatch dictionary - maps tool names to their execute functions
TOOL_HANDLERS: dict[
str, Callable[[JanaClient, dict[str, Any]], Awaitable[list[TextContent]]]
] = {
"get_air_quality": execute_air_quality,
"get_emissions": execute_emissions,
"find_nearby": execute_nearby,
"get_trends": execute_trends,
"get_data_summary": lambda client, _: execute_data_summary(client),
"get_system_health": lambda client, _: execute_system_health(client),
}
def create_mcp_server(settings: Settings | None = None) -> tuple[Server, JanaClient]:
"""
Create and configure the MCP server with all tools.
Args:
settings: Application settings. If None, uses default settings.
Returns:
Tuple of (MCP Server, Jana Client)
"""
# Create the Jana client
client = JanaClient(settings)
# Create and configure the server with this client
server = create_mcp_server_with_client(client)
return server, client
def create_mcp_server_with_client(client: JanaClient) -> Server:
"""
Create and configure the MCP server with an existing client.
This allows creating per-session MCP servers with user-specific
authenticated clients for the hosted service model.
Args:
client: Pre-configured Jana client (may have user-specific auth)
Returns:
Configured MCP Server
"""
# Create the MCP server
server = Server("jana-mcp-server")
# Register all tools with this client
_register_all_tools(server, client)
logger.info("MCP server created with all tools registered")
return server
def _register_all_tools(server: Server, client: JanaClient) -> None:
"""Register all available tools with the MCP server."""
# Register list_tools handler
@server.list_tools() # type: ignore[no-untyped-call,untyped-decorator]
async def handle_list_tools() -> list[Tool]:
"""Return all available tools."""
logger.debug("Listing %d tools", len(ALL_TOOLS))
return list(ALL_TOOLS)
# Register unified tool handler - routes to tool-specific execute functions
@server.call_tool() # type: ignore[untyped-decorator]
async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Route tool calls to appropriate handlers."""
logger.info("Tool call: %s", name)
# Route to appropriate handler using dispatch dictionary
handler = TOOL_HANDLERS.get(name)
if handler:
return await handler(client, arguments)
# Unknown tool
error_response = create_error_response(f"Unknown tool: {name}", "UNKNOWN_TOOL")
return [TextContent(type="text", text=serialize_response(error_response))]
logger.info("Registered %d tools", len(ALL_TOOLS))
