UniProt MCP Server

Instructions

Implementation Reference

• src/uniprot_mcp/server.py:145-171 (handler)

  The main tool handler for 'search_uniprot'. Decorated with @mcp.tool() for registration. Performs size validation, optional field optimization, calls the UniProt search API via search_json, and parses the response into SearchHit models.

  @mcp.tool()  # type: ignore[misc]
  async def search_uniprot(
      query: str,
      size: int = 10,
      reviewed_only: bool = False,
      fields: list[str] | None = None,
      sort: str | None = None,
      include_isoform: bool = False,
  ) -> list[SearchHit]:
      """Search UniProtKB and return curated hits."""
  
      size = max(1, min(size, 500))
      effective_fields = fields
      if fields is None and os.getenv("UNIPROT_ENABLE_FIELDS"):
          effective_fields = [FIELDS_SEARCH_LIGHT]
      async with new_client() as client:
          payload = await search_json(
              client,
              query=query,
              size=size,
              reviewed_only=reviewed_only,
              fields=effective_fields,
              sort=sort,
              include_isoform=include_isoform,
          )
      return parse_search_hits(payload)

• src/uniprot_mcp/models/domain.py:99-111 (schema)

  Pydantic model defining the structured output for each search hit returned by the tool.

  class SearchHit(BaseModel):
      """Result entry returned by UniProt search endpoints."""
  
      accession: str = Field(description="Primary accession identifier.")
      id: str | None = Field(default=None, description="UniProt entry name/ID.")
      reviewed: bool = Field(description="True for Swiss-Prot hits.")
      protein_name: str | None = Field(
          default=None, description="Recommended protein name when available."
      )
      organism: str | None = Field(
          default=None, description="Scientific name of the source organism."
      )

• src/uniprot_mcp/adapters/parsers.py:269-294 (helper)

  Helper function that parses the raw JSON search response from UniProt into a list of SearchHit Pydantic models.

  def parse_search_hits(js: dict[str, Any]) -> list[SearchHit]:
      """Convert a UniProt search response into SearchHit models."""
  
      hits: list[SearchHit] = []
      for result in js.get("results") or []:
          if not isinstance(result, dict):
              continue
          accession = result.get("primaryAccession") or result.get("accession")
          if not accession:
              continue
          organism_block = result.get("organism", {})
          organism_name = (
              organism_block.get("scientificName") if isinstance(organism_block, dict) else None
          )
          entry_type = result.get("entryType") or ""
          hits.append(
              SearchHit(
                  accession=str(accession),
                  id=result.get("uniProtkbId") or result.get("id"),
                  reviewed=str(entry_type).startswith("UniProtKB reviewed"),
                  protein_name=_extract_protein_name(result),
                  organism=organism_name,
              )
          )
      return hits

• src/uniprot_mcp/adapters/uniprot_client.py:255-298 (helper)

  Low-level HTTP client helper that executes the actual search request to UniProt's REST API (/uniprotkb/search) with retry logic, handles reviewed_only query modification, and returns raw JSON.

  async def search_json(
      client: httpx.AsyncClient,
      *,
      query: str,
      size: int,
      reviewed_only: bool = False,
      fields: Iterable[str] | None = None,
      sort: str | None = None,
      include_isoform: bool | None = None,
  ) -> dict[str, Any]:
      """Search UniProtKB and return the raw JSON response."""
  
      actual_query = query
      if reviewed_only and "reviewed:" not in query.lower():
          actual_query = f"({query}) AND reviewed:true"
  
      bounded_size = max(1, min(size, 500))
  
      params: dict[str, Any] = {
          "query": actual_query,
          "size": bounded_size,
      }
      if fields:
          params["fields"] = ",".join(fields)
      if sort:
          params["sort"] = sort
      if include_isoform is not None:
          params["includeIsoform"] = str(include_isoform).lower()
  
      async with _SEMAPHORE:
          response = await client.get(
              "/uniprotkb/search",
              params=params,
          )
      if response.status_code >= 400:
          if response.status_code in RETRYABLE_STATUS:
              response.raise_for_status()
          else:
              try:
                  response.raise_for_status()
              except httpx.HTTPStatusError as exc:
                  raise UniProtClientError(str(exc)) from exc
      return _ensure_dict(response.json())

• src/uniprot_mcp/server.py:145-145 (registration)

  The @mcp.tool() decorator registers the search_uniprot function as an MCP tool, using the function name as the tool name.

  @mcp.tool()  # type: ignore[misc]