src/tools/open_targets.py

import httpx
import time

GRAPHQL_URL = "https://api.platform.opentargets.org/api/v4/graphql"
ENSEMBL_URL = "https://rest.ensembl.org/lookup/symbol/homo_sapiens"


async def _resolve_ensembl_id(gene_symbol: str) -> str | None:
    """Resolve a gene symbol to an Ensembl gene ID."""
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get(
            f"{ENSEMBL_URL}/{gene_symbol}",
            headers={"Content-Type": "application/json"},
        )
        if r.status_code == 200:
            return r.json().get("id")
    return None


async def query_open_targets(gene_symbol: str) -> list[dict]:
    """Query Open Targets for drug-target associations."""
    ensembl_id = await _resolve_ensembl_id(gene_symbol)
    if not ensembl_id:
        return []

    query = """
    {
      target(ensemblId: "%s") {
        approvedName
        drugAndClinicalCandidates {
          rows {
            drug { name }
            maxClinicalStage
            diseases { diseaseFromSource }
          }
        }
      }
    }
    """ % ensembl_id

    async with httpx.AsyncClient(timeout=15) as client:
        r = await client.post(GRAPHQL_URL, json={"query": query})
        if r.status_code != 200:
            return []
        data = r.json().get("data", {}).get("target", {})

    results = []
    rows = data.get("drugAndClinicalCandidates", {}).get("rows", []) if data else []
    for row in rows[:5]:
        name = row.get("drug", {}).get("name", "")
        phase = row.get("maxClinicalStage", "")
        diseases = row.get("diseases", [])
        disease = diseases[0].get("diseaseFromSource", "") if diseases else ""
        results.append(
            {
                "url": f"https://platform.opentargets.org/target/{ensembl_id}",
                "snippet": f"{name} ({phase}) — associated with {disease}",
                "date_retrieved": int(time.time()),
                "source_name": "Open Targets",
            }
        )
    return results