EMS MCP Server

Instructions

Implementation Reference

• src/ems_mcp/tools/discovery.py:1054-1099 (handler)

  Main implementation of search_analytics tool - async function decorated with @mcp.tool that searches for time-series analytics by name, uses caching, and returns formatted results

  @mcp.tool
  async def search_analytics(
      ems_system_id: int,
      search_text: str,
      group_id: str | None = None,
      max_results: int = 50,
      show_ids: bool = False,
  ) -> str:
      """Search for time-series analytics by name (altitude, airspeed, etc.).
  
      You can pass analytic names directly to query_flight_analytics -- raw IDs
      are not needed. Use this tool to discover available analytic names.
  
      Args:
          ems_system_id: EMS system ID.
          search_text: Keyword to search for in analytic names.
          group_id: Optional analytic group ID to narrow search.
          max_results: Maximum results (default: 50).
          show_ids: If True, show full IDs inline.
  
      Returns:
          Matching analytics with names, types, units, and descriptions.
      """
      client = get_client()
  
      cache_key = make_cache_key(
          "analytics_search", ems_system_id, search_text.lower(), group_id or "all"
      )
      cached = await field_cache.get(cache_key)
      if cached is not None:
          logger.debug("Using cached analytics search: %s", cache_key)
          return _format_analytics_search_results(cached[:max_results], show_ids=show_ids)
  
      try:
          path = f"/api/v2/ems-systems/{ems_system_id}/analytics"
          params: dict[str, str] = {"text": search_text}
          if group_id:
              params["groupId"] = group_id
  
          analytics = await client.get(path, params=params)
          await field_cache.set(cache_key, analytics)
          return _format_analytics_search_results(analytics[:max_results], show_ids=show_ids)
      except EMSNotFoundError:
          return f"Error: EMS system {ems_system_id} not found. Use list_ems_systems to find valid system IDs."
      except EMSAPIError as e:
          return f"Error searching analytics: {e.message}"

• src/ems_mcp/tools/discovery.py:472-514 (helper)

  Helper function that formats analytics search results for display, optionally showing full IDs or assigning numbered references for easier use

  def _format_analytics_search_results(
      analytics: list[dict[str, Any]],
      show_ids: bool = False,
  ) -> str:
      """Format analytics search results for display.
  
      Args:
          analytics: List of analytic dicts from the API.
          show_ids: If True, show full IDs inline (backward compat).
              If False (default), assign numbered refs and hide IDs.
      """
      if not analytics:
          return "No analytics found matching the search criteria."
  
      lines = [f"Found {len(analytics)} analytic(s):"]
      for a in analytics:
          analytic_id = a.get("id", "?")
          analytic_name = a.get("name", "Unknown")
          analytic_type = a.get("type", "unknown")
          units = a.get("units")
          description = a.get("description")
  
          type_str = analytic_type
          if units:
              type_str = f"{analytic_type} ({units})"
  
          if show_ids:
              lines.append(f"\n  {analytic_name} [{type_str}]")
              if description:
                  lines.append(f"    {description}")
              lines.append(f"    ID: {analytic_id}")
          else:
              ref = _store_result(analytic_name, analytic_id, result_type="analytic")
              lines.append(f"\n  [{ref}] {analytic_name} [{type_str}]")
              if description:
                  lines.append(f"    {description}")
  
      if not show_ids:
          lines.append(
              "\nYou can pass analytic names directly to query_flight_analytics."
          )
  
      return "\n".join(lines)

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

  Export of search_analytics from discovery module to make it available for import in the package

  from ems_mcp.tools.discovery import (
      find_fields,
      get_field_info,
      get_result_id,
      list_databases,
      list_ems_systems,
      search_analytics,
  )

• src/ems_mcp/__init__.py:26-33 (registration)

  Import and export of search_analytics in the main package __init__ to make it available to external users

  from ems_mcp.tools import (
      find_fields,
      get_assets,
      get_field_info,
      list_databases,
      list_ems_systems,
      search_analytics,
  )

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

  Import statement that triggers registration of search_analytics with the FastMCP server via its @mcp.tool decorator

  # 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