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]