get_field_info
Retrieve field metadata including data type, units, and discrete value mappings for filtering and analysis in EMS flight databases.
Instructions
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| ems_system_id | Yes | ||
| database_id | Yes | ||
| field_id | Yes |
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}" - _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) - _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"] - _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", ]