mcp/targets.py

"""Open Targets GraphQL client – fetch gene ↔ disease ↔ drug associations.
     Docs: https://platform.opentargets.org/
     No API‑key required (public endpoint).
"""

import httpx, textwrap, asyncio
from typing import List, Dict

_OT_URL = "https://api.platform.opentargets.org/api/v4/graphql"

_QUERY_TEMPLATE = textwrap.dedent(
    """
    query Assoc($gene: String!, $size: Int!) {
      associations(geneSymbol: $gene, size: $size) {
        rows {
          score
          datatypeId
          datasourceId
          disease { id name }
          target  { id symbol }
        }
      }
    }
    """
)

async def fetch_ot_associations(gene_symbol: str, *, size: int = 30) -> List[Dict]:
    payload = {"query": _QUERY_TEMPLATE, "variables": {"gene": gene_symbol, "size": size}}
    async with httpx.AsyncClient(timeout=15) as client:
        resp = await client.post(_OT_URL, json=payload)
        resp.raise_for_status()
        return resp.json()["data"]["associations"]["rows"]