EMS MCP Server

Instructions

Implementation Reference

• src/ems_mcp/tools/discovery.py:897-932 (handler)

  Main implementation of list_databases tool. This async function navigates the database hierarchy, accepts ems_system_id and optional group_id parameters, fetches database groups from the EMS API with caching support, and returns formatted results showing databases and subgroups.

  @mcp.tool
  async def list_databases(
      ems_system_id: int,
      group_id: str | None = None,
  ) -> str:
      """Navigate the database hierarchy. Call without group_id for root level.
  
      The "FDW Flights" database (Flight Data Warehouse) contains flight records.
  
      Args:
          ems_system_id: EMS system ID (from list_ems_systems).
          group_id: Group ID to navigate into (omit for root).
  
      Returns:
          Databases and subgroups at the specified level.
      """
      client = get_client()
  
      cache_key = make_cache_key("database_group", ems_system_id, group_id or "root")
      cached = await database_cache.get(cache_key)
      if cached is not None:
          logger.debug("Using cached database group: %s", cache_key)
          return _format_database_group(cached)
  
      try:
          path = f"/api/v2/ems-systems/{ems_system_id}/database-groups"
          if group_id:
              path += f"?groupId={group_id}"
  
          group = await client.get(path)
          await database_cache.set(cache_key, group)
          return _format_database_group(group)
      except EMSNotFoundError:
          return f"Error: Database group not found. Verify ems_system_id={ems_system_id} is valid."
      except EMSAPIError as e:
          return f"Error listing databases: {e.message}"

• src/ems_mcp/tools/discovery.py:898-898 (registration)

  MCP tool registration decorator. The @mcp.tool decorator on line 898 registers list_databases as an available MCP tool with the FastMCP server instance.

  async def list_databases(

• src/ems_mcp/tools/discovery.py:311-354 (helper)

  Helper function that formats the database group response for display. This function creates a human-readable string showing group name, databases with their IDs and descriptions, and subgroups.

  def _format_database_group(group: dict[str, Any]) -> str:
      """Format database group response for display."""
      lines = []
  
      group_name = group.get("name", "Root")
      group_id = group.get("id", "[none]")
      lines.append(f"Group: {group_name} (ID: {group_id})")
  
      # Format databases
      databases = group.get("databases", [])
      if databases:
          lines.append(f"\nDatabases ({len(databases)}):")
          for db in databases:
              db_id = db.get("id", "?")
              # Handle both root level (name/description) and nested (pluralName/singularName)
              db_name = db.get("name") or db.get("pluralName") or db.get("singularName", "Unknown")
              desc = db.get("description", "")
              # Annotate entity-type-group IDs that require further navigation
              if "[entity-type-group]" in str(db_id):
                  note = " [NOTE: this is a group ID - navigate deeper with list_databases]"
                  lines.append(f"  - {db_name} (ID: {db_id}){note}")
              elif desc:
                  lines.append(f"  - {db_name}: {desc}")
              else:
                  lines.append(f"  - {db_name}")
  
      # Format subgroups
      groups = group.get("groups", [])
      if groups:
          lines.append(f"\nSubgroups ({len(groups)}):")
          for g in groups:
              g_id = g.get("id", "?")
              g_name = g.get("name", "Unknown")
              lines.append(f"  - {g_name} (ID: {g_id})")
  
      if not databases and not groups:
          lines.append("\n(Empty group)")
  
      if databases:
          lines.append(
              "\nUse database names directly in find_fields, query_database, etc."
          )
  
      return "\n".join(lines)

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

  Tool module exports. This section imports list_databases from the discovery module and re-exports it as part of the tools package's public API.

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

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

  Server module imports. The server imports ems_mcp.tools.discovery (line 129) which triggers the registration of list_databases via the @mcp.tool decorator when the module loads.

  # 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