EMS MCP Server

Instructions

Implementation Reference

• src/ems_mcp/tools/discovery.py:881-894 (handler)

  Main tool handler function decorated with @mcp.tool. Fetches EMS systems from the API endpoint and formats them for display.

  @mcp.tool
  async def list_ems_systems() -> str:
      """List available EMS systems. Start here to get system IDs for all other tools.
  
      Returns:
          EMS systems with IDs, names, and descriptions.
      """
      client = get_client()
  
      try:
          systems = await client.get("/api/v2/ems-systems")
          return _format_ems_systems(systems)
      except EMSAPIError as e:
          return f"Error listing EMS systems: {e.message}"

• src/ems_mcp/tools/discovery.py:294-308 (helper)

  Helper function that formats the list of EMS systems into a human-readable string output.

  def _format_ems_systems(systems: list[dict[str, Any]]) -> str:
      """Format EMS systems list for display."""
      if not systems:
          return "No EMS systems found."
  
      lines = [f"Found {len(systems)} EMS system(s):"]
      for sys in systems:
          name = sys.get("name", "Unknown")
          sys_id = sys.get("id", "?")
          desc = sys.get("description", "")
          if desc:
              lines.append(f"  - {name} (ID: {sys_id}): {desc}")
          else:
              lines.append(f"  - {name} (ID: {sys_id})")
      return "\n".join(lines)

• src/ems_mcp/api/models.py:118-124 (schema)

  Pydantic schema defining the structure of an EMS system object with id, name, and optional description fields.

  class EMSSystem(BaseModel):
      """EMS system information."""
  
      id: int
      name: str
      description: str | None = None

• src/ems_mcp/tools/__init__.py:13-38 (registration)

  Tool registration - imports list_ems_systems from discovery module and exports it in __all__ list for package-level availability.

  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/__init__.py:26-64 (registration)

  Top-level package registration - exports list_ems_systems in the main package's __all__ list for external access.

  from ems_mcp.tools import (
      find_fields,
      get_assets,
      get_field_info,
      list_databases,
      list_ems_systems,
      search_analytics,
  )
  
  __version__ = "0.2.0"
  
  __all__ = [
      # Server
      "mcp",
      "run",
      "get_client",
      # Client
      "EMSClient",
      # Auth
      "TokenManager",
      # Settings
      "EMSSettings",
      "get_settings",
      # Exceptions
      "AuthenticationError",
      "EMSAPIError",
      "EMSAuthorizationError",
      "EMSNotFoundError",
      "EMSRateLimitError",
      "EMSServerError",
      # Discovery Tools
      "list_ems_systems",
      "list_databases",
      "find_fields",
      "get_field_info",
      "search_analytics",
      # Asset Tools
      "get_assets",
  ]