EMS MCP Server

Instructions

Implementation Reference

• src/ems_mcp/tools/assets.py:143-172 (handler)

  Main handler function for ping_system tool. Decorated with @mcp.tool, it calls the EMS API ping endpoint and formats the response to show system ONLINE/OFFLINE status. Handles multiple response types (bool, str, dict) and API errors gracefully.

  @mcp.tool
  async def ping_system(ems_system_id: int) -> str:
      """Check if an EMS system is online and responsive.
  
      Args:
          ems_system_id: EMS system ID.
  
      Returns:
          System status.
      """
      client = get_client()
      try:
          path = f"/api/v2/ems-systems/{ems_system_id}/ping"
          response = await client.get(path)
          # Ping response can be a boolean, a string, or a dict with a message
          if isinstance(response, bool):
              status = "ONLINE" if response else "OFFLINE"
              return f"EMS System {ems_system_id} is {status}."
          elif isinstance(response, str):
              return f"EMS System {ems_system_id} is ONLINE. Response: {response}"
          elif isinstance(response, dict):
              message = response.get("message", "System is accessible")
              return f"EMS System {ems_system_id} is ONLINE. {message}"
          else:
              return f"EMS System {ems_system_id} is ONLINE."
      except EMSNotFoundError:
          return f"Error: EMS system {ems_system_id} not found."
      except EMSAPIError as e:
          return f"EMS System {ems_system_id} is OFFLINE or unreachable: {e.message}"

• src/ems_mcp/api/models.py:126-135 (schema)

  Pydantic schema definition for PingResponse model that defines the expected response structure from the EMS ping endpoint, including timestamp, server_time, status, and extra fields.

  class PingResponse(BaseModel):
      """Response from EMS system ping endpoint."""
  
      timestamp: datetime | None = None
      server_time: str | None = Field(default=None, alias="serverTime")
      status: str = "ok"
      extra: dict[str, Any] = Field(default_factory=dict)
  
      model_config = {"extra": "allow"}

• src/ems_mcp/tools/__init__.py:9-37 (registration)

  Module-level registration where ping_system is imported from assets.py and exported in __all__, making it available as part of the tools package.

  from ems_mcp.tools.assets import (
      get_assets,
      ping_system,
  )
  from ems_mcp.tools.discovery import (
      find_fields,
      get_field_info,
      get_result_id,
      list_databases,
      list_ems_systems,
      search_analytics,
  )
  from ems_mcp.tools.query import (
      query_database,
      query_flight_analytics,
  )
  
  __all__ = [
      "list_ems_systems",
      "list_databases",
      "find_fields",
      "get_field_info",
      "get_result_id",
      "search_analytics",
      "query_database",
      "query_flight_analytics",
      "get_assets",
      "ping_system",
  ]

• src/ems_mcp/server.py:126-132 (registration)

  Server initialization where ems_mcp.tools.assets module is imported, triggering the @mcp.tool decorator registration and making ping_system available to the FastMCP server.

  # Import tools and resources to register them with the mcp instance
  # This must happen after mcp is created
  import ems_mcp.tools.assets  # noqa: E402, F401
  import ems_mcp.tools.discovery  # noqa: E402, F401
  import ems_mcp.tools.query  # noqa: E402, F401
  import ems_mcp.prompts  # noqa: E402, F401
  import ems_mcp.resources  # noqa: E402, F401

• src/ems_mcp/resources.py:68-68 (helper)

  Documentation reference in workflow guide mentioning ping_system as a tool to check if a system is online.

  - ping_system checks if a system is online.