"""Framework-neutral MCP method and tool dispatch.""" from __future__ import annotations import json from dataclasses import dataclass from typing import Callable, Protocol from bot_bottle.gateway.egress.context import resolve_client_context from bot_bottle.gateway.egress.schema import route_to_yaml_dict from bot_bottle.gateway.policy_resolver import PolicyResolver from bot_bottle.supervisor import types as _sv class Request(Protocol): @property def method(self) -> str: ... @property def params(self) -> dict[str, object]: ... class MethodNotFoundError(Exception): """Raised when a JSON-RPC method has no MCP handler.""" Handler = Callable[[dict[str, object]], object] @dataclass(frozen=True) class Handlers: initialize: Handler tools_list: Handler list_routes: Handler check_proposal: Handler propose: Handler def dispatch(request: Request, handlers: Handlers) -> object: """Route one parsed request without depending on the HTTP server.""" if request.method == "initialize": return handlers.initialize(request.params) if request.method == "notifications/initialized": return None if request.method == "tools/list": return handlers.tools_list(request.params) if request.method != "tools/call": raise MethodNotFoundError(request.method) tool = request.params.get("name") if tool == _sv.TOOL_LIST_EGRESS_ROUTES: return handlers.list_routes(request.params) if tool == _sv.TOOL_CHECK_PROPOSAL: return handlers.check_proposal(request.params) return handlers.propose(request.params) def resolved_routes_payload( resolver: PolicyResolver, source_ip: str, identity_token: str, ) -> dict[str, object]: """Render the calling bottle's routes, failing closed to an empty list.""" config, _slug, _tokens = resolve_client_context( resolver, source_ip, identity_token, ) body = json.dumps( {"routes": [route_to_yaml_dict(route) for route in config.routes]}, indent=2, ) return {"content": [{"type": "text", "text": body}], "isError": False} __all__ = [ "Handlers", "MethodNotFoundError", "dispatch", "resolved_routes_payload", ]