import json import logging import contextvars from fastapi import APIRouter, Depends, Request, HTTPException from fastapi.responses import Response, JSONResponse from mcp.server import Server try: from mcp.server.streamable_http_manager import StreamableHTTPSessionManager except ImportError: from mcp.server.streamable_http import StreamableHTTPSessionManager from mcp.types import Tool, TextContent from app.auth import get_current_agent from app.scopes import has_scope_access, ALL_SCOPES from app.models import (get_rule, get_all_rules, get_all_ports, get_port, set_rule, get_agent_by_key, update_key_last_used, add_memory_entry, list_memory_entries) from app.git_sync import push_scope_markdown logger = logging.getLogger(__name__) router = APIRouter() mcp_server = Server("context-hub") current_agent_var: contextvars.ContextVar[str] = contextvars.ContextVar("current_agent", default="") @mcp_server.list_tools() async def handle_list_tools() -> list[Tool]: return [ Tool( name="get_rules", description="Get rules for a specific scope if authorized", inputSchema={ "type": "object", "properties": { "scope": {"type": "string", "description": "Scope name (infra, llm, nyora, perso, tt)"} }, "required": ["scope"] } ), Tool( name="get_ports", description="Get the full ports registry", inputSchema={ "type": "object", "properties": {} } ), Tool( name="search_rules", description="Search rules across all authorized scopes", inputSchema={ "type": "object", "properties": { "query": {"type": "string", "description": "Search term"} }, "required": ["query"] } ), Tool( name="get_agent_config", description="Get the compiled configuration for a specific agent based on its scopes", inputSchema={ "type": "object", "properties": { "name": {"type": "string", "description": "Agent name (e.g., GEMINI, HERMES_TT)"} }, "required": ["name"] } ), Tool( name="check_port", description="Check if a port is free or occupied", inputSchema={ "type": "object", "properties": { "port": {"type": "integer", "description": "Port number"} }, "required": ["port"] } ), Tool( name="get_my_context", description="Get all scopes authorized for this agent — call at START of session", inputSchema={"type": "object", "properties": {"agent": {"type": "string"}}, "required": ["agent"]} ), Tool( name="update_rule", description="Merge-update a scope with new key-value data — call at END of session", inputSchema={ "type": "object", "properties": { "scope": {"type": "string", "description": "Scope to update (infra/llm/nyora/perso/tt)"}, "agent": {"type": "string", "description": "Agent name for scope access check"}, "data": {"type": "object", "description": "Key-value pairs to merge into the scope"} }, "required": ["scope", "agent", "data"] } ), Tool( name="record_lesson", description=("Record a typed, durable memory entry (decision/constraint/best-practice/" "common-error/do-not-use) in the coding scope. Call after a meaningful " "decision, a repeated correction, or a rejected approach worth remembering."), inputSchema={ "type": "object", "properties": { "scope": {"type": "string", "description": "Scope, actuellement: coding"}, "project": {"type": "string", "description": "Nom du repo/projet, omettre si transverse"}, "type": {"type": "string", "enum": ["decision", "constraint", "best-practice", "common-error", "do-not-use"]}, "title": {"type": "string", "description": "Resume court, 1 ligne"}, "body": {"type": "string", "description": "1-3 phrases"}, "tags": {"type": "array", "items": {"type": "string"}} }, "required": ["scope", "type", "title", "body"] } ), Tool( name="get_context_pack", description=("Load only the memory entries relevant to the current task (selective " "loading, not the whole scope) — call before substantial coding work."), inputSchema={ "type": "object", "properties": { "scope": {"type": "string", "description": "Scope, actuellement: coding"}, "project": {"type": "string", "description": "Filtrer par projet ; omettre pour les entrees transverses"}, "type": {"type": "string", "enum": ["decision", "constraint", "best-practice", "common-error", "do-not-use"]} }, "required": ["scope"] } ) ] @mcp_server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]: # Autorisation liee a l'identite reelle de la connexion SSE (current_agent_var), # jamais a un champ "agent"/"name" declare par le client dans les arguments. caller_agent = current_agent_var.get() if name == "get_rules": scope = arguments.get("scope") if not has_scope_access(caller_agent, scope): return [TextContent(type="text", text=json.dumps({"error": f"Agent {caller_agent} forbidden on scope {scope}"}))] rule = await get_rule(scope) if not rule: return [TextContent(type="text", text=json.dumps({"error": "Scope not found"}))] return [TextContent(type="text", text=json.dumps(rule))] elif name == "get_ports": if not has_scope_access(caller_agent, "infra"): return [TextContent(type="text", text=json.dumps({"error": f"Agent {caller_agent} forbidden: requires infra scope"}))] ports = await get_all_ports() return [TextContent(type="text", text=json.dumps(ports))] elif name == "search_rules": query = arguments.get("query", "").lower() all_rules = await get_all_rules() results = [] for scope, content_ in all_rules.items(): if not has_scope_access(caller_agent, scope): continue if query in json.dumps(content_).lower(): results.append({"scope": scope, "content": content_}) return [TextContent(type="text", text=json.dumps(results))] elif name == "get_agent_config": all_rules = await get_all_rules() config = {} for scope in ALL_SCOPES: if has_scope_access(caller_agent, scope): config[scope] = all_rules.get(scope, {}) return [TextContent(type="text", text=json.dumps(config))] elif name == "check_port": if not has_scope_access(caller_agent, "infra"): return [TextContent(type="text", text=json.dumps({"error": f"Agent {caller_agent} forbidden: requires infra scope"}))] port = arguments.get("port") port_info = await get_port(port) if port_info: return [TextContent(type="text", text=json.dumps({"status": "occupied", "info": port_info}))] else: return [TextContent(type="text", text=json.dumps({"status": "free", "port": port}))] elif name == "get_my_context": all_rules = await get_all_rules() result = {} for scope in ALL_SCOPES: if has_scope_access(caller_agent, scope): result[scope] = all_rules.get(scope, {}) return [TextContent(type="text", text=json.dumps({"agent": caller_agent, "scopes": result}))] elif name == "update_rule": scope = arguments.get("scope") data = arguments.get("data", {}) if not has_scope_access(caller_agent, scope): return [TextContent(type="text", text=json.dumps({"error": f"Agent {caller_agent} forbidden on scope {scope}"}))] if not isinstance(data, dict): return [TextContent(type="text", text=json.dumps({"error": "data must be a JSON object"}))] existing = await get_rule(scope) or {} existing.update(data) await set_rule(scope, existing) return [TextContent(type="text", text=json.dumps({"status": "ok", "scope": scope, "updated_keys": list(data.keys())}))] elif name == "record_lesson": scope = arguments.get("scope") if not has_scope_access(caller_agent, scope): return [TextContent(type="text", text=json.dumps({"error": f"Agent {caller_agent} forbidden on scope {scope}"}))] type_ = arguments.get("type") title = arguments.get("title") body = arguments.get("body") if not (type_ and title and body): return [TextContent(type="text", text=json.dumps({"error": "type, title et body sont requis"}))] entry = await add_memory_entry( scope=scope, type_=type_, title=title, body=body, created_by=caller_agent, project=arguments.get("project"), tags=arguments.get("tags") ) git_result = await push_scope_markdown(scope) entry["_git"] = git_result return [TextContent(type="text", text=json.dumps(entry))] elif name == "get_context_pack": scope = arguments.get("scope") if not has_scope_access(caller_agent, scope): return [TextContent(type="text", text=json.dumps({"error": f"Agent {caller_agent} forbidden on scope {scope}"}))] entries = await list_memory_entries( scope=scope, project=arguments.get("project"), type_=arguments.get("type"), status="active" ) return [TextContent(type="text", text=json.dumps(entries))] return [TextContent(type="text", text=json.dumps({"error": "Unknown tool"}))] # Streamable HTTP Integration (mcp SDK v1.x, transport recommande depuis la spec 2025-03-26). # Remplace l ancien transport SSE (SseServerTransport) qui exigeait un session_id # etabli via un GET prealable -- les clients MCP modernes (Gemini/Antigravity inclus) # postent directement le JSON-RPC initialize sans cette poignee de main, d ou les 400 # Bad Request observes avec l ancienne implementation. # L auth X-API-Key reste verifiee a la main avant de deleguer a session_manager.handle_request, # meme principe qu avant : current_agent_var est positionne pour la duree de la requete, # lu ensuite par les handlers call_tool/list_tools via caller_agent. session_manager = StreamableHTTPSessionManager( app=mcp_server, event_store=None, json_response=True, stateless=True, ) async def _mcp_asgi_app(scope, receive, send): if scope["type"] != "http": return headers = dict(scope.get("headers") or []) api_key = headers.get(b"x-api-key", b"").decode() agent = await get_agent_by_key(api_key) if api_key else None if not agent: response = JSONResponse({"detail": "Invalid or missing API Key"}, status_code=401) await response(scope, receive, send) return await update_key_last_used(api_key) token = current_agent_var.set(agent) try: await session_manager.handle_request(scope, receive, send) finally: current_agent_var.reset(token) router.mount("/mcp", _mcp_asgi_app)