UniProt MCP Server

Instructions

Implementation Reference

• src/uniprot_mcp/server.py:244-270 (handler)

  The main handler function for the 'map_ids' tool. It filters input IDs, submits a mapping job to UniProt via the client, polls for completion using _poll_mapping_job, and parses the result using parse_mapping_result. The @mcp.tool() decorator registers it as an MCP tool.

  @mcp.tool()  # type: ignore[misc]
  async def map_ids(
      from_db: str,
      to_db: str,
      ids: list[str],
      ctx: Context[ServerSession, None] | None = None,
  ) -> MappingResult:
      """Map identifiers between UniProt-supported namespaces."""
  
      filtered_ids = [identifier for identifier in ids if identifier]
      if not filtered_ids:
          return MappingResult(from_db=from_db, to_db=to_db, results={})
  
      async with new_client() as client:
          job_id = await start_id_mapping(client, from_db=from_db, to_db=to_db, ids=filtered_ids)
  
      if ctx is not None:
          await ctx.info(
              f"Submitted UniProt ID mapping job ({from_db}->{to_db}) for {len(filtered_ids)} IDs."
          )
  
      payload = await _poll_mapping_job(job_id, ctx=ctx)
      if ctx is not None:
          await ctx.info(
              f"Completed UniProt ID mapping job ({from_db}->{to_db}) for {len(filtered_ids)} IDs."
          )
      return parse_mapping_result(payload, from_db=from_db, to_db=to_db)

• src/uniprot_mcp/models/domain.py:113-122 (schema)

  Pydantic BaseModel defining the structured output schema for the map_ids tool, including from_db, to_db, and a dictionary of results mapping input IDs to lists of target IDs.

  class MappingResult(BaseModel):
      """Outcome of UniProt ID mapping operations."""
  
      from_db: str = Field(description="Source identifier namespace.")
      to_db: str = Field(description="Target identifier namespace.")
      results: dict[str, list[str]] = Field(
          default_factory=dict,
          description="Mapping from input IDs to resolved identifiers (empty list for no match).",
      )

• src/uniprot_mcp/adapters/parsers.py:296-350 (helper)

  Helper function to parse the raw JSON response from UniProt ID mapping into the MappingResult model, handling various response formats, failed IDs, and deduplication.

  def parse_mapping_result(
      js: dict[str, Any],
      *,
      from_db: str,
      to_db: str,
  ) -> MappingResult:
      """Convert an ID mapping response into MappingResult."""
  
      mappings: dict[str, list[str]] = {}
  
      def register_result(source: str | None, targets: Iterable[Any]) -> None:
          if not source:
              return
          values: list[str] = []
          for target in targets:
              if isinstance(target, dict):
                  candidate = target.get("id") or target.get("identifier") or target.get("value")
                  if candidate:
                      values.append(str(candidate))
              elif target is not None:
                  values.append(str(target))
          if source not in mappings:
              mappings[source] = []
          mappings[source].extend(values)
  
      for item in js.get("results") or []:
          if not isinstance(item, dict):
              continue
          source = item.get("from") or item.get("fromId")
          to_value = item.get("to") or item.get("toId") or item.get("mappedTo")
          if isinstance(to_value, list):
              register_result(source, to_value)
          elif to_value is not None:
              register_result(source, [to_value])
          else:
              register_result(source, [])
  
      # Some responses return an explicit mapping dictionary
      for source, value in (js.get("mappedResults") or {}).items():
          if isinstance(value, list):
              register_result(source, value)
          else:
              register_result(source, [value])
  
      # Ensure failed IDs are tracked with empty lists
      for failed in js.get("failedIds") or []:
          if failed not in mappings:
              mappings[failed] = []
  
      # Normalise ordering and remove duplicates per ID
      for key, values in mappings.items():
          deduped = list(dict.fromkeys(values))
          mappings[key] = deduped
  
      return MappingResult(from_db=from_db, to_db=to_db, results=mappings)

• src/uniprot_mcp/server.py:198-222 (helper)

  Helper function to poll the UniProt ID mapping job status until completion or timeout, with progress reporting via MCP context.

  async def _poll_mapping_job(
      job_id: str,
      *,
      ctx: Context[ServerSession, None] | None = None,
  ) -> dict[str, Any]:
      """Poll the UniProt mapping job until completion or timeout."""
  
      elapsed = 0.0
      async with new_client() as client:
          while elapsed < MAPPING_MAX_WAIT:
              status = await get_mapping_status(client, job_id)
              if _mapping_is_complete(status):
                  results = await get_mapping_results(client, job_id)
                  return cast(dict[str, Any], results)
              elapsed += MAPPING_POLL_INTERVAL
              if ctx is not None:
                  progress = min(1.0, elapsed / MAPPING_MAX_WAIT)
                  await ctx.report_progress(
                      progress=progress,
                      total=1.0,
                      message="Polling UniProt ID mapping job",
                  )
              await asyncio.sleep(MAPPING_POLL_INTERVAL)
          raise UniProtClientError("ID mapping timed out waiting for completion.")