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)