Agent MCP Gateway

Instructions

Implementation Reference

• src/gateway.py:526-589 (handler)

  The main asynchronous handler function that implements the logic for the get_gateway_status tool. It collects diagnostic information from various components (reload status, policy state, servers, config paths) and returns a structured JSON response using the GatewayStatusResponse model.
  async def get_gateway_status( agent_id: Annotated[Optional[str], "Your agent identifier (leave empty if not provided to you)"] = None ) -> dict: """Get gateway status, configuration state, and hot reload diagnostics. NOTE: Only available when debug mode is enabled.""" # Defensive check (middleware should have resolved agent_id) if agent_id is None: raise ToolError("Internal error: agent_id not resolved by middleware") # Get reload status if available reload_status = None if _get_reload_status_fn: try: reload_status = _get_reload_status_fn() # Convert datetime objects to ISO strings for JSON serialization if reload_status: for config_type in ["mcp_config", "gateway_rules"]: if config_type in reload_status: for key in ["last_attempt", "last_success"]: if reload_status[config_type].get(key): reload_status[config_type][key] = reload_status[config_type][key].isoformat() except Exception: reload_status = {"error": "Failed to retrieve reload status"} # Get PolicyEngine state policy_state = {} if _policy_engine: try: policy_state = { "total_agents": len(_policy_engine.agents), "agent_ids": list(_policy_engine.agents.keys()), "defaults": _policy_engine.defaults, } except Exception: policy_state = {"error": "Failed to retrieve policy state"} # Get available servers from ProxyManager (reflects hot-reload changes) available_servers = [] if _proxy_manager: try: available_servers = _proxy_manager.get_all_servers() except Exception: available_servers = [] # Get config file paths from src/config.py config_paths = {} try: from src.config import get_stored_config_paths mcp_path, rules_path = get_stored_config_paths() config_paths = { "mcp_config": mcp_path, "gateway_rules": rules_path, } except Exception: config_paths = {"error": "Failed to retrieve config paths"} return GatewayStatusResponse( reload_status=reload_status, policy_state=policy_state, available_servers=available_servers, config_paths=config_paths, message="Gateway is operational. Check reload_status for hot reload health." ).model_dump()
• src/gateway.py:47-54 (schema)

  Pydantic BaseModel defining the structured output schema for the get_gateway_status tool response, including fields for reload status, policy state, available servers, config paths, and a status message.
  class GatewayStatusResponse(BaseModel): """Response from get_gateway_status (debug tool).""" reload_status: Annotated[Optional[dict], Field(description="Hot reload history with timestamps and errors")] policy_state: Annotated[dict, Field(description="Policy engine configuration (agent count, defaults)")] available_servers: Annotated[list[str], Field(description="All configured server names")] config_paths: Annotated[dict, Field(description="File paths to gateway configuration")] message: Annotated[str, Field(description="Summary status message")]
• src/gateway.py:151-162 (registration)

  The _register_debug_tools function conditionally registers the get_gateway_status tool with the FastMCP gateway instance when debug mode is enabled. This is called from initialize_gateway if debug_mode=True.
  def _register_debug_tools(): """Register debug-only tools when debug mode is enabled. This function is called by initialize_gateway() when debug_mode=True. It registers additional diagnostic tools that should only be available in debug/development environments. """ # Register get_gateway_status tool # Note: The function itself is always defined (for testing), but only # registered as a gateway tool when debug mode is enabled gateway.tool(get_gateway_status)