EMS MCP Server

Instructions

Implementation Reference

• src/ems_mcp/tools/discovery.py:1102-1139 (handler)

  Main handler function for get_result_id tool. Registered with @mcp.tool decorator. Takes a list of reference numbers and returns the full opaque IDs for each result stored in the result store. The function is marked as DEPRECATED since query_database and get_field_info now accept [N] references directly.

  @mcp.tool
  async def get_result_id(
      result_numbers: list[int],
  ) -> str:
      """DEPRECATED: query_database and get_field_info now accept [N] reference
      numbers and field names directly. This tool is no longer needed in the
      standard workflow.
  
      Retrieve full opaque IDs for numbered [N] references from search results.
  
      Args:
          result_numbers: Reference numbers from search results (e.g., [1, 3, 5]).
  
      Returns:
          The name and full ID for each requested result.
      """
      if not result_numbers:
          return "Error: result_numbers cannot be empty."
  
      lines: list[str] = []
      not_found: list[int] = []
  
      for ref in result_numbers:
          entry = _get_stored_result(ref)
          if entry is not None:
              type_label = f" ({entry['type']})" if entry.get("type") else ""
              lines.append(f"[{ref}] {entry['name']}{type_label}")
              lines.append(f"  ID: {entry['id']}")
          else:
              not_found.append(ref)
  
      if not_found:
          lines.append(
              f"\nNot found: {not_found}. These may have been evicted or never existed. "
              "Re-run the search to get fresh references."
          )
  
      return "\n".join(lines) if lines else "No results found for the given reference numbers."

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

  Tool registration via @mcp.tool decorator. This decorator registers the get_result_id function with the FastMCP server instance imported from ems_mcp.server.

  @mcp.tool

• src/ems_mcp/tools/discovery.py:61-70 (helper)

  Helper function _get_stored_result that retrieves a stored result from the result store by reference number. Used by get_result_id to look up results that were previously stored from search operations.

  def _get_stored_result(ref: int) -> dict[str, str] | None:
      """Look up a stored result by reference number.
  
      Args:
          ref: The reference number returned by ``_store_result``.
  
      Returns:
          Dict with ``name`` and ``id`` keys, or ``None`` if not found.
      """
      return _result_store.get(ref)

• src/ems_mcp/tools/discovery.py:31-58 (helper)

  Helper function _store_result that stores a result and returns its reference number. This populates the _result_store that get_result_id queries. Implements eviction logic when the store exceeds maximum capacity.

  def _store_result(name: str, result_id: str, result_type: str = "field") -> int:
      """Store a result and return its reference number.
  
      Entries accumulate across searches so the agent can reference results
      from any prior search within the session. When the store exceeds
      ``_STORE_MAX_SIZE``, the oldest entries are evicted.
  
      Args:
          name: Human-readable name of the result.
          result_id: The full opaque ID string.
          result_type: Type of result: ``"field"`` or ``"analytic"``.
  
      Returns:
          The reference number assigned to this result.
      """
      global _next_ref  # noqa: PLW0603
  
      ref = _next_ref
      _next_ref += 1
      _result_store[ref] = {"name": name, "id": result_id, "type": result_type}
  
      # Evict oldest entries when over capacity
      if len(_result_store) > _STORE_MAX_SIZE:
          oldest_keys = sorted(_result_store.keys())[: len(_result_store) - _STORE_MAX_SIZE]
          for k in oldest_keys:
              del _result_store[k]
  
      return ref