EMS MCP Server

Instructions

Implementation Reference

• src/ems_mcp/tools/discovery.py:995-1051 (handler)

  Main get_field_info tool handler with @mcp.tool decorator. Implements field metadata retrieval including type, units, and discrete value mappings. Resolves database and field references, uses caching, and formats output for display.

  @mcp.tool
  async def get_field_info(
      ems_system_id: int,
      database_id: str,
      field_id: str | int,
  ) -> str:
      """Get field metadata including type, units, and discrete value mappings.
  
      Essential for discrete fields: shows numeric code-to-label mappings needed
      for filtering. String labels in filters are auto-resolved, but use this to
      verify available values.
  
      Args:
          ems_system_id: EMS system ID.
          database_id: Database ID or name (e.g. "FDW Flights").
          field_id: Field reference: [N] number from find_fields, field name
              (e.g. "Takeoff Airport Name"), or bracket-encoded ID.
  
      Returns:
          Field details with discrete value mappings if applicable.
      """
      # Resolve database name -> ID
      try:
          database_id = await _resolve_database_id(database_id, ems_system_id)
      except ValueError as e:
          return f"Error resolving database: {e}"
  
      # Resolve field reference -> opaque ID
      try:
          field_id = await _resolve_field_id(field_id, ems_system_id, database_id)
      except (ValueError, EMSAPIError) as e:
          return f"Error resolving field: {e}"
  
      client = get_client()
  
      cache_key = make_cache_key("field_info", ems_system_id, database_id, field_id)
      cached = await field_cache.get(cache_key)
      if cached is not None:
          logger.debug("Using cached field info: %s", cache_key)
          return _format_field_info(cached)
  
      try:
          encoded_field_id = urllib.parse.quote(field_id, safe="")
          path = (
              f"/api/v2/ems-systems/{ems_system_id}/databases/{database_id}/fields/{encoded_field_id}"
          )
  
          field = await client.get(path)
          await field_cache.set(cache_key, field)
          return _format_field_info(field)
      except EMSNotFoundError:
          return (
              "Error: Field not found. Verify field_id is correct. "
              "Use find_fields to find valid field IDs."
          )
      except EMSAPIError as e:
          return f"Error getting field info: {e.message}"

• src/ems_mcp/tools/discovery.py:431-469 (helper)

  _format_field_info helper function that formats field metadata for display. Handles field name, type, units, description, and discrete values (with limiting for large sets). Returns formatted string output.

  def _format_field_info(field: dict[str, Any]) -> str:
      """Format detailed field information for display."""
      lines = []
  
      field_name = field.get("name", "Unknown")
      field_id = field.get("id", "?")
      field_type = field.get("type", "unknown")
  
      lines.append(f"Field: {field_name}")
      lines.append(f"Type: {field_type}")
  
      units = field.get("units")
      if units:
          lines.append(f"Units: {units}")
  
      description = field.get("description")
      if description:
          lines.append(f"Description: {description}")
  
      lines.append(f"\nField ID: {field_id}")
  
      # Handle discrete values
      discrete_values = field.get("discreteValues")
      if discrete_values:
          # Normalize dict format {"code": "label"} to list format [{"value": code, "label": label}]
          if isinstance(discrete_values, dict):
              discrete_values = [{"value": k, "label": v} for k, v in discrete_values.items()]
  
          lines.append(f"\nDiscrete Values ({len(discrete_values)}):")
          # Limit display for large value sets
          display_count = min(len(discrete_values), 50)
          for dv in discrete_values[:display_count]:
              value = dv.get("value", "?")
              label = dv.get("label", "Unknown")
              lines.append(f"  {value}: {label}")
          if len(discrete_values) > display_count:
              lines.append(f"  ... and {len(discrete_values) - display_count} more values")
  
      return "\n".join(lines)

• src/ems_mcp/tools/discovery.py:80-170 (helper)

  _resolve_field_id helper that resolves field references to opaque field IDs. Supports integer references (result store lookup), bracket-encoded IDs, and human-readable names via API search.

  async def _resolve_field_id(
      field_ref: str | int,
      ems_system_id: int,
      database_id: str,
  ) -> str:
      """Resolve a field reference to an opaque field ID.
  
      Resolution order:
      1. Integer or digit string -> look up in result store
      2. Bracket-encoded string (starts with ``[``) -> pass through
      3. Human-readable name -> search field API, exact or single match
  
      Args:
          field_ref: A result store reference number, bracket-encoded ID, or
              human-readable field name.
          ems_system_id: The EMS system ID for API lookups.
          database_id: The database ID for API lookups.
  
      Returns:
          The resolved opaque field ID string.
  
      Raises:
          ValueError: If the reference cannot be resolved.
      """
      # 1. Integer or digit string -> result store lookup
      if isinstance(field_ref, int) or (isinstance(field_ref, str) and field_ref.strip().isdigit()):
          ref_num = int(field_ref) if isinstance(field_ref, str) else field_ref
          entry = _get_stored_result(ref_num)
          if entry is not None:
              if entry.get("type") == "analytic":
                  raise ValueError(
                      f"Reference [{ref_num}] ('{entry['name']}') is an analytic parameter, "
                      "not a database field. Use it with query_flight_analytics, "
                      "or use find_fields to find database field references."
                  )
              return entry["id"]
          raise ValueError(
              f"Reference [{ref_num}] not found in result store. "
              "Re-run find_fields to get fresh references."
          )
  
      if not isinstance(field_ref, str) or not field_ref.strip():
          raise ValueError(f"Invalid field reference: {field_ref!r}")
  
      field_ref = field_ref.strip()
  
      # 2. Bracket-encoded string -> pass through
      if field_ref.startswith("["):
          return field_ref
  
      # 3. Human-readable name -> search via API
      cache_key = make_cache_key("field_resolve", ems_system_id, database_id, field_ref.lower())
      cached = await field_cache.get(cache_key)
      if cached is not None:
          return cached
  
      client = get_client()
  
      # Entity-type databases don't support the field search endpoint (405);
      # fall back to BFS traversal of field groups.
      if _is_entity_type_database(database_id):
          matches, _ = await _recursive_field_search(
              client, ems_system_id, database_id,
              search_text=field_ref,
              max_depth=10, max_results=50, max_groups=50,
          )
          search_results = matches
      else:
          path = f"/api/v2/ems-systems/{ems_system_id}/databases/{database_id}/fields"
          params = {"text": field_ref}
          search_results = await client.get(path, params=params)
  
      if not search_results:
          raise ValueError(
              f"Field not found: '{field_ref}'. "
              "Use find_fields to discover valid field names."
          )
  
      # Try exact name match (case-insensitive)
      exact_matches = [
          f for f in search_results
          if f.get("name", "").lower() == field_ref.lower()
      ]
      if len(exact_matches) == 1:
          resolved_id = exact_matches[0]["id"]
          await field_cache.set(cache_key, resolved_id)
          return resolved_id
  
      # Single result total -> use it
      if len(search_results) == 1:
          resolved_id = search_results[0]["id"]

• src/ems_mcp/tools/discovery.py:184-280 (helper)

  _resolve_database_id helper that resolves database names to opaque database IDs. Supports bracket-encoded IDs and human-readable names via cached database groups API lookup.

  async def _resolve_database_id(
      database_ref: str,
      ems_system_id: int,
  ) -> str:
      """Resolve a database name to an opaque database ID.
  
      Resolution order:
      1. Bracket-encoded string (starts with ``[``) -> pass through
      2. Human-readable name -> look up via database groups API
  
      The first call fetches root database groups + one level deep and caches
      the full name-to-ID mapping.
  
      Args:
          database_ref: A bracket-encoded database ID or human-readable name.
          ems_system_id: The EMS system ID for API lookups.
  
      Returns:
          The resolved opaque database ID string.
  
      Raises:
          ValueError: If the name cannot be resolved.
      """
      if not database_ref or not database_ref.strip():
          raise ValueError("database_id cannot be empty.")
  
      database_ref = database_ref.strip()
  
      # 1. Bracket-encoded -> pass through
      if database_ref.startswith("["):
          return database_ref
  
      # 2. Name -> look up in cached mapping
      cache_key = make_cache_key("database_name_map", ems_system_id)
      name_map: dict[str, str] | None = await database_cache.get(cache_key)
  
      if name_map is None:
          # Build the name -> ID mapping from root + one level of subgroups
          client = get_client()
          name_map = {}
  
          try:
              root = await client.get(f"/api/v2/ems-systems/{ems_system_id}/database-groups")
          except (EMSAPIError, EMSNotFoundError) as e:
              raise ValueError(f"Failed to fetch database groups: {e}") from e
  
          # Collect databases at root
          for db in root.get("databases", []):
              db_id = db.get("id", "")
              for name_key in ("name", "pluralName", "singularName"):
                  db_name = db.get(name_key)
                  if db_name:
                      name_map[db_name.lower()] = db_id
  
          # Fetch one level of subgroups
          for group in root.get("groups", []):
              group_id = group.get("id")
              if not group_id:
                  continue
              try:
                  sub = await client.get(
                      f"/api/v2/ems-systems/{ems_system_id}/database-groups?groupId={group_id}"
                  )
                  for db in sub.get("databases", []):
                      db_id = db.get("id", "")
                      for name_key in ("name", "pluralName", "singularName"):
                          db_name = db.get(name_key)
                          if db_name:
                              name_map[db_name.lower()] = db_id
              except (EMSAPIError, EMSNotFoundError):
                  continue
  
          await database_cache.set(cache_key, name_map)
  
      # Case-insensitive lookup
      resolved = name_map.get(database_ref.lower())
      if resolved is not None:
          return resolved
  
      # Not found
      available = sorted(set(name_map.keys()))[:10]
      raise ValueError(
          f"Database not found: '{database_ref}'. "
          f"Available databases include: {', '.join(available)}"
          f"{'...' if len(name_map) > 10 else ''}. "
          "Use list_databases to browse available databases."
      )
  
  
  def _is_entity_type_database(database_id: str) -> bool:
      """Check if a database ID is an entity-type database.
  
      Entity-type databases (containing ``[entity-type]`` but not
      ``[entity-type-group]``) don't support the field search endpoint.
      Fields must be discovered via browse mode or deep search instead.
  
      Args:

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

  Tool registration and export from tools package. get_field_info is imported from discovery module and listed in __all__ for public API exposure.

  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",
  ]