get_gateway_status

Instructions

Implementation Reference

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

  The actual implementation of the get_gateway_status tool. It collects reload status (with datetime→ISO conversion), policy engine state (agent counts/IDs/defaults), available servers from ProxyManager, config file paths via get_stored_config_paths(), and returns a GatewayStatusResponse dict.

  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-53 (schema)

  GatewayStatusResponse Pydantic model defining the response schema for get_gateway_status: reload_status, policy_state, available_servers, config_paths, and 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-161 (registration)

  Registration of get_gateway_status as a gateway tool via gateway.tool(get_gateway_status). Only called when 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)

• src/gateway.py:69-102 (helper)

  initialize_gateway() accepts debug_mode flag and conditionally calls _register_debug_tools() to expose the get_gateway_status tool.

  def initialize_gateway(
      policy_engine: PolicyEngine,
      mcp_config: dict,
      proxy_manager: ProxyManager | None = None,
      check_config_changes_fn: Any = None,
      get_reload_status_fn: Any = None,
      default_agent_id: str | None = None,
      debug_mode: bool = False
  ):
      """Initialize gateway with policy engine, MCP config, and proxy manager.
  
      This must be called before the gateway starts accepting requests.
  
      Args:
          policy_engine: PolicyEngine instance for access control
          mcp_config: MCP servers configuration dictionary
          proxy_manager: Optional ProxyManager instance (required for get_server_tools)
          check_config_changes_fn: Optional function to check for config changes (fallback mechanism)
          get_reload_status_fn: Optional function to get reload status for diagnostics
          default_agent_id: Optional default agent ID from GATEWAY_DEFAULT_AGENT env var for fallback chain
          debug_mode: Whether debug mode is enabled (exposes get_gateway_status tool)
      """
      global _policy_engine, _mcp_config, _proxy_manager, _check_config_changes_fn, _get_reload_status_fn, _default_agent_id, _debug_mode
      _policy_engine = policy_engine
      _mcp_config = mcp_config
      _proxy_manager = proxy_manager
      _check_config_changes_fn = check_config_changes_fn
      _get_reload_status_fn = get_reload_status_fn
      _default_agent_id = default_agent_id
      _debug_mode = debug_mode
  
      # Conditionally register debug tools based on debug mode
      if debug_mode:
          _register_debug_tools()