search_clinical_trials

Instructions

Implementation Reference

• aria_mcp_server/tools.py:264-294 (handler)

  Core handler function that queries the ClinicalTrials.gov v2 API with condition, status, intervention, and max_results parameters. Returns a list of parsed trial dicts.

  def search_clinical_trials(
      condition: str,
      status: str = "RECRUITING",
      intervention: str = "",
      max_results: int = 5,
  ) -> list[dict]:
      """Search ClinicalTrials.gov v2 API. No API key required."""
      condition = (condition or "").strip()
      if not condition:
          return []
      max_results = max(1, min(max_results, 20))
      params: dict[str, Any] = {
          "format": "json",
          "query.cond": condition,
          "pageSize": max_results,
          "countTotal": "true",
      }
      if intervention:
          params["query.intr"] = intervention.strip()
      if status and status.upper() != "ALL":
          params["filter.overallStatus"] = status.upper()
  
      try:
          r = requests.get(CT_BASE, params=params, timeout=15)
          r.raise_for_status()
          data = r.json()
      except Exception as e:
          raise RuntimeError(f"ClinicalTrials.gov search failed: {e}") from e
  
      studies = data.get("studies") or []
      return [t for t in (_parse_trial(s) for s in studies) if t]

• aria_mcp_server/tools.py:188-261 (schema)

  Parses the raw API response (study dict) into a structured dict with fields like nct_id, title, status, phase, conditions, interventions, eligibility criteria, locations, etc.

  def _parse_trial(study: dict) -> dict | None:
      try:
          proto = study.get("protocolSection") or {}
          id_mod = proto.get("identificationModule") or {}
          status_mod = proto.get("statusModule") or {}
          design_mod = proto.get("designModule") or {}
          desc_mod = proto.get("descriptionModule") or {}
          elig_mod = proto.get("eligibilityModule") or {}
          sponsor_mod = proto.get("sponsorCollaboratorsModule") or {}
          conditions_mod = proto.get("conditionsModule") or {}
          interventions_mod = proto.get("armsInterventionsModule") or {}
          contacts_mod = proto.get("contactsLocationsModule") or {}
          outcomes_mod = proto.get("outcomesModule") or {}
  
          primary_outcomes = outcomes_mod.get("primaryOutcomes") or []
          primary_outcome = primary_outcomes[0].get("measure", "") if primary_outcomes else ""
          if len(primary_outcome) > 300:
            primary_outcome = primary_outcome[:297] + "..."
  
          secondary_outcomes_list = outcomes_mod.get("secondaryOutcomes") or []
          secondary_outcome_strs = [
            o.get("measure", "") for o in secondary_outcomes_list[:3] if isinstance(o, dict)
          ]
          secondary_outcome = "; ".join(secondary_outcome_strs)
          if len(secondary_outcome) > 300:
            secondary_outcome = secondary_outcome[:297] + "..."
  
          nct_id = _ct_get(id_mod, "nctId")
          if not nct_id:
              return None
  
          interventions = interventions_mod.get("interventions") or []
          intervention_names = "; ".join(
              _ct_get(i, "name") for i in interventions if isinstance(i, dict)
          )
          locations = contacts_mod.get("locations") or []
          location_strs = []
          for loc in locations[:3]:
              if not isinstance(loc, dict):
                  continue
              parts = [_ct_get(loc, "facility"), _ct_get(loc, "city"), _ct_get(loc, "country")]
              loc_str = ", ".join(p for p in parts if p)
              if loc_str:
                  location_strs.append(loc_str)
  
          criteria = _ct_get(elig_mod, "eligibilityCriteria")
          if len(criteria) > 600:
              criteria = criteria[:597] + "..."
  
          summary = _ct_get(desc_mod, "briefSummary")
          if len(summary) > 400:
              summary = summary[:397] + "..."
  
          return {
              "nct_id": nct_id,
              "title": _ct_get(id_mod, "briefTitle"),
              "status": _ct_get(status_mod, "overallStatus"),
              "phase": "; ".join(design_mod.get("phases") or []),
              "conditions": "; ".join(conditions_mod.get("conditions") or []),
              "interventions": intervention_names,
              "brief_summary": summary,
              "eligibility_criteria": criteria,
              "min_age": _ct_get(elig_mod, "minimumAge"),
              "max_age": _ct_get(elig_mod, "maximumAge"),
              "sex": _ct_get(elig_mod, "sex"),
              "sponsor": _ct_get(sponsor_mod, "leadSponsor", "name"),
              "start_date": _ct_get(status_mod, "startDateStruct", "date"),
              "locations": location_strs,
              "url": f"https://clinicaltrials.gov/study/{nct_id}",
              "primary_outcome": primary_outcome,
              "secondary_outcomes": secondary_outcome,
          }
      except (KeyError, TypeError, AttributeError):
          return None

• aria_mcp_server/server.py:83-103 (registration)

  Registration of the 'search_clinical_trials' tool with the FastMCP server, including description and output schema.

  @mcp.tool(
      description=(
          "Search ClinicalTrials.gov for clinical studies. "
          "Read-only operation. No authentication required. "
          "Uses ClinicalTrials.gov v2 public API (no rate limit documented). "
          "Returns up to 10 results per call. No pagination. "
          "Returns 'No clinical trials found.' if no results match. "
          "Use for: active trials, recruiting studies, eligibility criteria, "
          "phase information, sponsor details, and trial locations."
      ),
      output_schema={
          "type": "object",
          "properties": {
              "result": {
                  "type": "string",
                  "description": "Formatted list of trials with NCT ID, title, phase, status, sponsor, conditions, interventions, and eligibility criteria. Returns 'no results' message if nothing found."
              }
          },
          "required": ["result"]
      }
  )

• aria_mcp_server/server.py:105-143 (handler)

  MCP tool wrapper function that accepts user-facing parameters with Annotated type hints, delegates to the core handler in tools.py, and formats results as a string.

  def search_clinical_trials(
      condition: Annotated[str, "Disease or condition e.g. 'pediatric epilepsy', 'lung cancer'"],
      status: Annotated[str, "Trial status: RECRUITING, COMPLETED, or ALL"] = "RECRUITING",
      intervention: Annotated[str, "Optional drug or intervention name to narrow results"] = "",
      max_results: Annotated[int, "Number of trials to return, between 1 and 10"] = 5,
  ) -> str:
      """
      Search ClinicalTrials.gov for clinical studies.
  
      Use for: active trials, recruiting studies, trial eligibility criteria,
      phase information, sponsor details, and trial locations.
  
      Args:
          condition: Disease or condition (e.g. "pediatric epilepsy", "lung cancer")
          status: Trial status — RECRUITING, COMPLETED, or ALL (default: RECRUITING)
          intervention: Optional drug or intervention name to narrow results
          max_results: Number of trials to return (1-10, default 5)
      
      Returns:
          Formatted string with NCT ID, title, phase, status, sponsor, conditions,
          interventions, and eligibility criteria for each trial.
          Returns a "no results" message if nothing is found.
          Handles API errors gracefully with descriptive error messages.
      
      Notes:
          - status defaults to RECRUITING if not specified
          - intervention is optional and narrows results when provided
          - max_results is clamped to 1-10 regardless of input
          - Requires no API key; uses ClinicalTrials.gov v2 public API
      """
      from aria_mcp_server.tools import search_clinical_trials as _search, format_trials_for_claude as _fmt
      max_results = max(1, min(max_results, 10))
      trials = _search(
          condition=condition,
          status=status,
          intervention=intervention,
          max_results=max_results,
      )
      return _fmt(trials)

• aria_mcp_server/tools.py:297-323 (helper)

  Formats the list of trial dicts into a human-readable string for display to the user.

  def format_trials_for_claude(trials: list[dict]) -> str:
      """Format ClinicalTrials.gov results as readable text."""
      if not trials:
          return "No clinical trials found matching those criteria."
      lines = []
      for i, t in enumerate(trials, 1):
          locations_str = "; ".join(t.get("locations") or []) or "N/A"
          lines.append("\n".join([
              f"[Trial {i}]",
              f"NCT ID: {t.get('nct_id') or 'N/A'}",
              f"Title: {t.get('title') or 'N/A'}",
              f"Status: {t.get('status') or 'N/A'}",
              f"Phase: {t.get('phase') or 'N/A'}",
              f"Condition(s): {t.get('conditions') or 'N/A'}",
              f"Intervention(s): {t.get('interventions') or 'N/A'}",
              f"Sponsor: {t.get('sponsor') or 'N/A'}",
              f"Age Range: {t.get('min_age') or 'N/A'} – {t.get('max_age') or 'N/A'}",
              f"Sex: {t.get('sex') or 'N/A'}",
              f"Locations: {locations_str}",
              f"Summary: {t.get('brief_summary') or 'N/A'}",
              f"Primary Outcome: {t.get('primary_outcome') or 'N/A'}",
              f"Secondary Outcomes: {t.get('secondary_outcomes') or 'N/A'}",
              f"Eligibility: {t.get('eligibility_criteria') or 'N/A'}",
              f"URL: {t.get('url') or 'N/A'}",
              "",
          ]))
      return "\n".join(lines).strip()