BioContextAI Knowledgebase MCP

Instructions

Implementation Reference

• src/biocontext_kb/core/clinicaltrials/_get_studies_by_intervention.py:9-116 (handler)

  The handler function get_studies_by_intervention decorated with @core_mcp.tool(). Queries ClinicalTrials.gov API for studies matching the intervention (e.g., drug name), with optional filters for condition, phase, status, intervention_type. Processes response to include summary breakdowns (status, phase, conditions, sponsors). Tool name prefixed to 'bc_get_studies_by_intervention' via server import.

  @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/clinicaltrials/_get_studies_by_intervention.py:10-31 (schema)

  Pydantic-based input schema defined via Annotated[ ] and Field( ) descriptions for the tool parameters.

  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]:

• src/biocontext_kb/core/clinicaltrials/__init__.py:3-3 (registration)

  Explicit import of the tool handler function in the clinicaltrials package __init__.py, which executes the @tool() decorator registration.

  from ._get_studies_by_intervention import get_studies_by_intervention

• src/biocontext_kb/core/__init__.py:7-7 (registration)

  Imports all clinicaltrials tools in core package __init__.py, importing the function and triggering its registration on core_mcp.

  from .clinicaltrials import *

• src/biocontext_kb/app.py:35-39 (registration)

  Main app.py imports core_mcp into the root FastMCP server with prefix slugify('BC')='bc', making the tool available as 'bc_get_studies_by_intervention'.

  for mcp in [core_mcp, *(await get_openapi_mcps())]:
      await mcp_app.import_server(
          mcp,
          slugify(mcp.name),
      )