EMS MCP Server

Instructions

Implementation Reference

• src/ems_mcp/tools/assets.py:90-141 (handler)

  Main async handler function for get_assets tool. Retrieves reference data (fleets, aircraft, airports, flight phases) from EMS API based on ems_system_id and asset_type parameters. Supports optional fleet_id filter for aircraft queries. Returns formatted string output.

  async def get_assets(
      ems_system_id: int,
      asset_type: Literal["fleets", "aircraft", "airports", "flight_phases"],
      fleet_id: int | None = None,
  ) -> str:
      """Get reference data: fleets, aircraft, airports, or flight phases.
  
      Args:
          ems_system_id: EMS system ID (from list_ems_systems).
          asset_type: Type of assets to retrieve.
          fleet_id: Filter aircraft by fleet ID (only for asset_type="aircraft").
  
      Returns:
          Formatted list of the requested asset type.
      """
      client = get_client()
  
      try:
          if asset_type == "fleets":
              path = f"/api/v2/ems-systems/{ems_system_id}/assets/fleets"
              data = await client.get(path)
              return _format_fleets(data)
  
          elif asset_type == "aircraft":
              path = f"/api/v2/ems-systems/{ems_system_id}/assets/aircraft"
              params: dict[str, Any] = {}
              if fleet_id is not None:
                  params["fleetId"] = fleet_id
              data = await client.get(path, params=params)
              return _format_aircraft(data)
  
          elif asset_type == "airports":
              path = f"/api/v2/ems-systems/{ems_system_id}/assets/airports"
              data = await client.get(path)
              return _format_airports(data)
  
          elif asset_type == "flight_phases":
              path = f"/api/v2/ems-systems/{ems_system_id}/assets/flight-phases"
              data = await client.get(path)
              return _format_flight_phases(data)
  
          else:
              return (
                  f"Error: Unknown asset_type '{asset_type}'. "
                  "Valid types: fleets, aircraft, airports, flight_phases."
              )
  
      except EMSNotFoundError:
          return f"Error: EMS system {ems_system_id} not found."
      except EMSAPIError as e:
          return f"Error getting {asset_type}: {e.message}"

• src/ems_mcp/tools/assets.py:90-94 (schema)

  Function signature defines the input/output schema: ems_system_id (int), asset_type (Literal with 4 valid values), optional fleet_id (int | None), returns string. The Literal type restricts asset_type to valid values: 'fleets', 'aircraft', 'airports', 'flight_phases'.

  async def get_assets(
      ems_system_id: int,
      asset_type: Literal["fleets", "aircraft", "airports", "flight_phases"],
      fleet_id: int | None = None,
  ) -> str:

• src/ems_mcp/tools/assets.py:89-89 (registration)

  The @mcp.tool decorator registers the get_assets function with the FastMCP server instance, making it available as an MCP tool.

  @mcp.tool

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

  Module import statement that triggers tool registration. Importing ems_mcp.tools.assets causes the @mcp.tool decorated functions to be registered with the FastMCP server instance.

  import ems_mcp.tools.assets  # noqa: E402, F401

• src/ems_mcp/tools/assets.py:15-86 (helper)

  Four helper formatting functions that convert API response data into human-readable formatted strings: _format_fleets, _format_aircraft, _format_flight_phases, _format_airports. Each handles empty lists and formats multiple items with appropriate details.

  def _format_fleets(fleets: list[dict[str, Any]]) -> str:
      """Format fleets list for display."""
      if not fleets:
          return "No fleets found."
  
      lines = [f"Found {len(fleets)} fleet(s):"]
      for f in fleets:
          name = f.get("name", "Unknown")
          fleet_id = f.get("id", "?")
          desc = f.get("description", "")
          if desc:
              lines.append(f"  - {name} (ID: {fleet_id}): {desc}")
          else:
              lines.append(f"  - {name} (ID: {fleet_id})")
      return "\n".join(lines)
  
  
  def _format_aircraft(aircraft: list[dict[str, Any]]) -> str:
      """Format aircraft list for display."""
      if not aircraft:
          return "No aircraft found."
  
      lines = [f"Found {len(aircraft)} aircraft:"]
      for a in aircraft:
          name = a.get("name", "Unknown")
          aircraft_id = a.get("id", "?")
          fleet_name = a.get("fleetName", "Unknown")
          lines.append(f"  - {name} (ID: {aircraft_id}) [Fleet: {fleet_name}]")
      return "\n".join(lines)
  
  
  def _format_flight_phases(phases: list[dict[str, Any]]) -> str:
      """Format flight phases list for display."""
      if not phases:
          return "No flight phases found."
  
      lines = [f"Found {len(phases)} flight phase(s):"]
      for p in phases:
          name = p.get("name", "Unknown")
          phase_id = p.get("id", "?")
          desc = p.get("description", "")
          if desc:
              lines.append(f"  - {name} (ID: {phase_id}): {desc}")
          else:
              lines.append(f"  - {name} (ID: {phase_id})")
      return "\n".join(lines)
  
  
  def _format_airports(airports: list[dict[str, Any]]) -> str:
      """Format airports list for display."""
      if not airports:
          return "No airports found."
  
      lines = [f"Found {len(airports)} airport(s):"]
      for a in airports:
          icao = a.get("codeIcao", "????")
          iata = a.get("codeIata")
          name = a.get("name", "Unknown")
          city = a.get("city", "")
          country = a.get("country", "")
  
          location = ", ".join(part for part in (city, country) if part)
          id_str = f" (ID: {a.get('id', '?')})"
  
          codes = f"{icao}/{iata}" if iata else icao
  
          line = f"  - {codes}: {name}"
          if location:
              line += f" [{location}]"
          line += id_str
          lines.append(line)
      return "\n".join(lines)