bc_search_drugs_fda

Instructions

Implementation Reference

• src/biocontext_kb/core/openfda/_search_drugs.py:9-112 (handler)

  The main handler function for the 'search_drugs_fda' tool. Decorated with @core_mcp.tool(), it builds FDA Drugs@FDA API queries from optional search parameters (brand_name, generic_name, active_ingredient, sponsor_name, etc.), sends requests to https://api.fda.gov/drug/drugsfda.json, and returns JSON results.

  @core_mcp.tool()
  def search_drugs_fda(
      brand_name: Annotated[Optional[str], Field(description="Brand or trade name (e.g., 'Tylenol')")] = None,
      generic_name: Annotated[Optional[str], Field(description="Generic name (e.g., 'acetaminophen')")] = None,
      active_ingredient: Annotated[Optional[str], Field(description="Active ingredient name")] = None,
      sponsor_name: Annotated[Optional[str], Field(description="Company/sponsor name")] = None,
      application_number: Annotated[
          Optional[str], Field(description="FDA application number (NDA, ANDA, or BLA)")
      ] = None,
      marketing_status: Annotated[
          Optional[str],
          Field(
              description="Marketing status: 'Prescription', 'Over-the-counter', 'Discontinued', or 'None (Tentative Approval)'"
          ),
      ] = None,
      dosage_form: Annotated[
          Optional[str], Field(description="Dosage form (e.g., 'TABLET', 'INJECTION', 'CAPSULE')")
      ] = None,
      route: Annotated[
          Optional[str], Field(description="Route of administration (e.g., 'ORAL', 'INJECTION', 'TOPICAL')")
      ] = None,
      search_type: Annotated[str, Field(description="'and' for all terms must match, 'or' for any term matches")] = "or",
      sort_by: Annotated[
          Optional[str], Field(description="Field to sort by (e.g., 'sponsor_name', 'application_number')")
      ] = None,
      limit: Annotated[int, Field(description="Number of results to return", ge=1, le=1000)] = 25,
      skip: Annotated[int, Field(description="Number of results to skip for pagination", ge=0, le=25000)] = 0,
  ) -> dict:
      """Search FDA Drugs@FDA database for approved drug products. Supports multiple search criteria.
  
      Returns:
          dict: Results array with drug products including application numbers, sponsors, products array or error message.
      """
      # Ensure at least one search parameter is provided
      search_params = [
          brand_name,
          generic_name,
          active_ingredient,
          sponsor_name,
          application_number,
          marketing_status,
          dosage_form,
          route,
      ]
      if not any(search_params):
          return {"error": "At least one search parameter must be provided"}
  
      # Build query components - using correct schema field paths
      query_parts = []
  
      if brand_name:
          # Search in both openfda.brand_name and products.brand_name arrays
          query_parts.append(f"(openfda.brand_name:{brand_name} OR products.brand_name:{brand_name})")
  
      if generic_name:
          # openfda.generic_name is an array
          query_parts.append(f"openfda.generic_name:{generic_name}")
  
      if active_ingredient:
          # products.active_ingredients.name
          query_parts.append(f"products.active_ingredients.name:{active_ingredient}")
  
      if sponsor_name:
          query_parts.append(f"sponsor_name:{sponsor_name}")
  
      if application_number:
          query_parts.append(f"application_number.exact:{application_number}")
  
      if marketing_status:
          # Map user-friendly terms to API values - products.marketing_status
          status_mapping = {
              "prescription": "1",
              "discontinued": "2",
              "none (tentative approval)": "3",
              "over-the-counter": "4",
          }
          status_value = status_mapping.get(marketing_status.lower(), marketing_status)
          query_parts.append(f"products.marketing_status:{status_value}")
  
      if dosage_form:
          query_parts.append(f"products.dosage_form:{dosage_form}")
  
      if route:
          query_parts.append(f"products.route:{route}")
  
      # Join query parts based on search type
      query = " AND ".join(query_parts) if search_type.lower() == "and" else " OR ".join(query_parts)
  
      # Build URL parameters for proper encoding
      params = {"search": query, "limit": limit, "skip": skip}
  
      # Add sorting if specified
      if sort_by:
          params["sort"] = f"{sort_by}:desc"
  
      # Build the complete URL
      base_url = "https://api.fda.gov/drug/drugsfda.json"
  
      try:
          response = requests.get(base_url, params=params)  # type: ignore
          response.raise_for_status()
          return response.json()
      except requests.exceptions.RequestException as e:
          return {"error": f"Failed to fetch FDA drug data: {e!s}"}

• src/biocontext_kb/core/openfda/_search_drugs.py:10-36 (schema)

  Pydantic-annotated type hints defining the input schema for search_drugs_fda, including brand_name, generic_name, active_ingredient, sponsor_name, application_number, marketing_status, dosage_form, route, search_type, sort_by, limit, and skip with descriptions and constraints.

  def search_drugs_fda(
      brand_name: Annotated[Optional[str], Field(description="Brand or trade name (e.g., 'Tylenol')")] = None,
      generic_name: Annotated[Optional[str], Field(description="Generic name (e.g., 'acetaminophen')")] = None,
      active_ingredient: Annotated[Optional[str], Field(description="Active ingredient name")] = None,
      sponsor_name: Annotated[Optional[str], Field(description="Company/sponsor name")] = None,
      application_number: Annotated[
          Optional[str], Field(description="FDA application number (NDA, ANDA, or BLA)")
      ] = None,
      marketing_status: Annotated[
          Optional[str],
          Field(
              description="Marketing status: 'Prescription', 'Over-the-counter', 'Discontinued', or 'None (Tentative Approval)'"
          ),
      ] = None,
      dosage_form: Annotated[
          Optional[str], Field(description="Dosage form (e.g., 'TABLET', 'INJECTION', 'CAPSULE')")
      ] = None,
      route: Annotated[
          Optional[str], Field(description="Route of administration (e.g., 'ORAL', 'INJECTION', 'TOPICAL')")
      ] = None,
      search_type: Annotated[str, Field(description="'and' for all terms must match, 'or' for any term matches")] = "or",
      sort_by: Annotated[
          Optional[str], Field(description="Field to sort by (e.g., 'sponsor_name', 'application_number')")
      ] = None,
      limit: Annotated[int, Field(description="Number of results to return", ge=1, le=1000)] = 25,
      skip: Annotated[int, Field(description="Number of results to skip for pagination", ge=0, le=25000)] = 0,
  ) -> dict:

• src/biocontext_kb/core/openfda/__init__.py:8-19 (registration)

  The function is exported from the openfda package's __init__.py, imported from ._search_drugs and listed in __all__.

  from ._search_drugs import search_drugs_fda
  
  __all__ = [
      "count_drugs_by_field",
      "get_available_pharmacologic_classes",
      "get_drug_by_application_number",
      "get_drug_label_info",
      "get_drug_statistics",
      "get_generic_equivalents",
      "search_drugs_by_therapeutic_class",
      "search_drugs_fda",
  ]

• src/biocontext_kb/core/_server.py:1-6 (registration)

  The core_mcp FastMCP server instance used as the decorator (@core_mcp.tool()) to register search_drugs_fda as an MCP tool.

  from fastmcp import FastMCP
  
  core_mcp = FastMCP(  # type: ignore
      "BC",
      instructions="Provides access to biomedical knowledge bases.",
  )

• tests/core/test_openfda.py:9-163 (helper)

  Test cases for the search_drugs_fda tool, covering brand name, generic name, sponsor name, and no-parameter scenarios by calling the tool via the MCP client.

  async def test_search_drugs_fda_by_brand_name():
      """Test the search_drugs_fda function with brand name search."""
      async with Client(core_mcp) as client:
          result_text = await client.call_tool("search_drugs_fda", {"brand_name": "Tylenol", "limit": 5})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          assert "error" not in result or "results" in result
  
  
  async def test_search_drugs_fda_by_generic_name():
      """Test the search_drugs_fda function with generic name search."""
      async with Client(core_mcp) as client:
          result_text = await client.call_tool("search_drugs_fda", {"generic_name": "acetaminophen", "limit": 5})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          assert "error" not in result or "results" in result
  
  
  async def test_search_drugs_fda_by_sponsor():
      """Test the search_drugs_fda function with sponsor search."""
      async with Client(core_mcp) as client:
          result_text = await client.call_tool("search_drugs_fda", {"sponsor_name": "Johnson & Johnson", "limit": 5})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          assert "error" not in result or "results" in result
  
  
  async def test_get_drug_by_application_number():
      """Test the get_drug_by_application_number function."""
      async with Client(core_mcp) as client:
          result_text = await client.call_tool("get_drug_by_application_number", {"application_number": "NDA021436"})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          assert "error" not in result or "results" in result
  
  
  async def test_get_drug_label_info():
      """Test the get_drug_label_info function."""
      async with Client(core_mcp) as client:
          result_text = await client.call_tool("get_drug_label_info", {"brand_name": "Aspirin"})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          assert "error" not in result or "results" in result
  
  
  async def test_count_drugs_by_field():
      """Test the count_drugs_by_field function."""
      async with Client(core_mcp) as client:
          result_text = await client.call_tool("count_drugs_by_field", {"field": "sponsor_name", "limit": 10})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          assert "error" not in result or "results" in result
  
  
  async def test_get_drug_statistics():
      """Test the get_drug_statistics function."""
      async with Client(core_mcp) as client:
          result_text = await client.call_tool("get_drug_statistics", {})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          # Should have multiple statistics sections
          if "error" not in result:
              assert any(
                  key in result for key in ["top_sponsors", "dosage_forms", "administration_routes", "marketing_statuses"]
              )
  
  
  async def test_get_available_pharmacologic_classes():
      """Test the get_available_pharmacologic_classes function."""
      async with Client(core_mcp) as client:
          result_text = await client.call_tool("get_available_pharmacologic_classes", {"class_type": "epc", "limit": 10})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          if "error" not in result:
              assert "available_classes" in result
              assert "class_type" in result
              assert "total_found" in result
          else:
              # API might return an error, which is acceptable
              assert "error" in result
  
  
  async def test_search_drugs_by_therapeutic_class():
      """Test the search_drugs_by_therapeutic_class function with a real FDA class term."""
      async with Client(core_mcp) as client:
          # First get available classes to use a real term
          classes_result = (
              await client.call_tool("get_available_pharmacologic_classes", {"class_type": "epc", "limit": 5})
          ).data
  
          assert isinstance(classes_result, dict)
  
          if "error" not in classes_result and classes_result.get("available_classes"):
              # Use the first available class
              first_class = classes_result["available_classes"][0]["term"]
  
              result_text = await client.call_tool(
                  "search_drugs_by_therapeutic_class", {"therapeutic_class": first_class, "class_type": "epc", "limit": 5}
              )
              result = json.loads(result_text.content[0].text)
  
              assert isinstance(result, dict)
              assert "error" not in result or "results" in result
          else:
              # If we can't get classes, test with a known term that might exist
              result_text = await client.call_tool(
                  "search_drugs_by_therapeutic_class",
                  {"therapeutic_class": "Nonsteroidal Anti-inflammatory Drug [EPC]", "class_type": "epc", "limit": 5},
              )
              result = json.loads(result_text.content[0].text)
  
              assert isinstance(result, dict)
              # This might return an error or results, both are acceptable
  
  
  async def test_get_generic_equivalents():
      """Test the get_generic_equivalents function."""
      async with Client(core_mcp) as client:
          result_text = await client.call_tool("get_generic_equivalents", {"brand_name": "Advil"})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          # Should either return results or an error message
          assert "error" in result or "brand_drug" in result
  
  
  async def test_search_drugs_fda_no_params():
      """Test the search_drugs_fda function with no search parameters."""
      async with Client(core_mcp) as client:
          result_text = await client.call_tool("search_drugs_fda", {})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          assert "error" in result
          assert "at least one search parameter" in result["error"].lower()
  
  
  async def test_count_drugs_by_field_invalid_field():
      """Test the count_drugs_by_field function with potentially invalid field."""
      async with Client(core_mcp) as client:
          # This should work, but might return an error from the API
          result_text = await client.call_tool("count_drugs_by_field", {"field": "invalid_field_name", "limit": 10})
          result = json.loads(result_text.content[0].text)
  
          assert isinstance(result, dict)
          # Should either return results or an error