bc_get_studies_by_intervention
Search clinical trials by drug or intervention name, with filters for condition, phase, status, and type. Retrieve results with summary statistics for biomedical research.
Instructions
Search for clinical trials by drug or intervention name.
This function helps biomedical researchers find clinical trials testing specific drugs, therapies, or treatments, with optional filters for condition and phase.
Args: intervention (str): Drug, therapy, or treatment name to search for. condition (str, optional): Medical condition to filter by. phase (str, optional): Clinical trial phase to filter by. status (str, optional): Study status filter (default: "ALL"). intervention_type (str, optional): Type of intervention filter (default: "ALL"). page_size (int): Number of results to return (default: 50, max: 1000). sort (str): Sort order for results (default: most recently updated).
Returns: dict: Study search results with summary statistics or error message
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| condition | No | Medical condition to filter by (e.g., 'cancer', 'diabetes') | |
| intervention | Yes | Drug, therapy, or treatment name (e.g., 'aspirin', 'pembrolizumab', 'radiation therapy') | |
| intervention_type | No | Type of intervention: 'DRUG', 'BIOLOGICAL', 'DEVICE', 'PROCEDURE', 'RADIATION', 'BEHAVIORAL', 'ALL' | ALL |
| page_size | No | Number of results to return | |
| phase | No | Clinical trial phase: 'PHASE1', 'PHASE2', 'PHASE3', 'PHASE4', 'EARLY_PHASE1' | |
| sort | No | Sort order: 'LastUpdatePostDate:desc', 'StudyFirstPostDate:desc', 'EnrollmentCount:desc' | LastUpdatePostDate:desc |
| status | No | Study status: 'RECRUITING', 'ACTIVE_NOT_RECRUITING', 'COMPLETED', 'ALL' | ALL |
Implementation Reference
- The core handler function `get_studies_by_intervention` decorated with `@core_mcp.tool()`, which queries the ClinicalTrials.gov API for studies matching the intervention (e.g., drug name), with optional filters for condition, phase, status, intervention type, pagination, and sorting. Includes summary statistics like breakdowns by status, phase, conditions, and sponsors. The FastMCP server named 'BC' likely prefixes the tool name to 'bc_get_studies_by_intervention'.from typing import Annotated, Any, Dict, Optional, Union import requests from pydantic import Field from biocontext_kb.core._server import core_mcp @core_mcp.tool() def get_studies_by_intervention( intervention: Annotated[ str, Field(description="Drug/therapy name (e.g., 'aspirin', 'pembrolizumab', 'radiation')"), ], condition: Annotated[Optional[str], Field(description="Medical condition filter (e.g., 'cancer')")] = None, phase: Annotated[ Optional[str], Field(description="'PHASE1', 'PHASE2', 'PHASE3', 'PHASE4', or 'EARLY_PHASE1'") ] = None, status: Annotated[ Optional[str], Field(description="'RECRUITING', 'ACTIVE_NOT_RECRUITING', 'COMPLETED', or 'ALL'") ] = "ALL", intervention_type: Annotated[ Optional[str], Field(description="'DRUG', 'BIOLOGICAL', 'DEVICE', 'PROCEDURE', 'RADIATION', 'BEHAVIORAL', or 'ALL'"), ] = "ALL", page_size: Annotated[int, Field(description="Results per page (1-1000)", ge=1, le=1000)] = 50, sort: Annotated[ str, Field(description="'LastUpdatePostDate:desc', 'StudyFirstPostDate:desc', or 'EnrollmentCount:desc'"), ] = "LastUpdatePostDate:desc", ) -> Union[Dict[str, Any], dict]: """Search trials by intervention with condition and phase filters. Returns paginated results with breakdowns. Returns: dict: Studies list with summary containing intervention searched, total studies, status/phase breakdowns, top conditions/sponsors or error message. """ if not intervention: return {"error": "Intervention name must be provided"} # Build query components query_parts = [f"AREA[InterventionName]{intervention}"] if condition: query_parts.append(f"AREA[ConditionSearch]{condition}") if phase: query_parts.append(f"AREA[Phase]{phase}") if status and status != "ALL": query_parts.append(f"AREA[OverallStatus]{status}") if intervention_type and intervention_type != "ALL": query_parts.append(f"AREA[InterventionType]{intervention_type}") # Join query parts with AND query = " AND ".join(query_parts) url = f"https://clinicaltrials.gov/api/v2/studies?query.term={query}&pageSize={page_size}&sort={sort}&format=json" try: response = requests.get(url) response.raise_for_status() data = response.json() # Add summary statistics if "studies" in data: total_studies = data.get("totalCount", len(data["studies"])) # Count studies by various attributes status_counts: dict[str, int] = {} phase_counts: dict[str, int] = {} condition_counts: dict[str, int] = {} sponsor_counts: dict[str, int] = {} for study in data["studies"]: # Extract status status_module = study.get("protocolSection", {}).get("statusModule", {}) study_status = status_module.get("overallStatus", "Unknown") status_counts[study_status] = status_counts.get(study_status, 0) + 1 # Extract phase design_module = study.get("protocolSection", {}).get("designModule", {}) phases = design_module.get("phases", []) if phases: for phase_item in phases: phase_counts[phase_item] = phase_counts.get(phase_item, 0) + 1 else: phase_counts["N/A"] = phase_counts.get("N/A", 0) + 1 # Extract primary conditions conditions_module = study.get("protocolSection", {}).get("conditionsModule", {}) conditions = conditions_module.get("conditions", []) if conditions: for cond in conditions[:3]: # Limit to first 3 conditions condition_counts[cond] = condition_counts.get(cond, 0) + 1 # Extract lead sponsor sponsor_module = study.get("protocolSection", {}).get("sponsorCollaboratorsModule", {}) lead_sponsor = sponsor_module.get("leadSponsor", {}).get("name", "Unknown") sponsor_counts[lead_sponsor] = sponsor_counts.get(lead_sponsor, 0) + 1 # Add summary to response data["summary"] = { "intervention_searched": intervention, "total_studies": total_studies, "studies_returned": len(data["studies"]), "status_breakdown": status_counts, "phase_breakdown": phase_counts, "top_conditions": dict(sorted(condition_counts.items(), key=lambda x: x[1], reverse=True)[:10]), "top_sponsors": dict(sorted(sponsor_counts.items(), key=lambda x: x[1], reverse=True)[:10]), } return data except requests.exceptions.RequestException as e: return {"error": f"Failed to fetch studies by intervention: {e!s}"}
- src/biocontext_kb/core/_server.py:1-7 (registration)Definition of the `core_mcp` FastMCP server instance with name 'BC'. Tool handler functions decorated with `@core_mcp.tool()` are automatically registered here, with tool names prefixed by 'bc_' (e.g., 'bc_get_studies_by_intervention'). This server is then imported into the main app.from fastmcp import FastMCP core_mcp = FastMCP( # type: ignore "BC", instructions="Provides access to biomedical knowledge bases.", )
- Pydantic-based input schema defined via Annotated types and Field descriptions for parameters: intervention (required str), condition/phase/status/intervention_type (optional str), page_size (int 1-1000), sort (str). Output is dict with studies list and summary stats.@core_mcp.tool() def get_studies_by_intervention( intervention: Annotated[ str, Field(description="Drug/therapy name (e.g., 'aspirin', 'pembrolizumab', 'radiation')"), ], condition: Annotated[Optional[str], Field(description="Medical condition filter (e.g., 'cancer')")] = None, phase: Annotated[ Optional[str], Field(description="'PHASE1', 'PHASE2', 'PHASE3', 'PHASE4', or 'EARLY_PHASE1'") ] = None, status: Annotated[ Optional[str], Field(description="'RECRUITING', 'ACTIVE_NOT_RECRUITING', 'COMPLETED', or 'ALL'") ] = "ALL", intervention_type: Annotated[ Optional[str], Field(description="'DRUG', 'BIOLOGICAL', 'DEVICE', 'PROCEDURE', 'RADIATION', 'BEHAVIORAL', or 'ALL'"), ] = "ALL", page_size: Annotated[int, Field(description="Results per page (1-1000)", ge=1, le=1000)] = 50, sort: Annotated[ str, Field(description="'LastUpdatePostDate:desc', 'StudyFirstPostDate:desc', or 'EnrollmentCount:desc'"), ] = "LastUpdatePostDate:desc", ) -> Union[Dict[str, Any], dict]: