bc_query_kegg

Instructions

Implementation Reference

• src/biocontext_kb/core/kegg/_query_kegg.py:223-269 (handler)

  The main handler function for the 'query_kegg' tool. It accepts KEGG operation, database, query, entries, etc., builds a KeggConfig, validates it, and executes the KEGG API query.

  @core_mcp.tool()
  def query_kegg(
      operation: Annotated[KeggOperation, Field(description="info, list, find, get, conv, link, or ddi")],
      database: Annotated[
          Optional[Union[KeggDatabase, KeggOutsideDb, str]],
          Field(description="pathway, compound, genes, organism code (hsa, mmu, etc.), or other DB"),
      ] = None,
      target_db: Annotated[
          Optional[Union[KeggDatabase, KeggOutsideDb, str]],
          Field(description="Target DB for conversion/linking operations"),
      ] = None,
      source_db: Annotated[
          Optional[Union[KeggDatabase, KeggOutsideDb, str]],
          Field(description="Source DB for conversion/linking operations"),
      ] = None,
      query: Annotated[Optional[str], Field(description="Query string for FIND/LIST, or organism code for LIST")] = None,
      option: Annotated[
          Optional[Union[KeggOption, KeggFindOption, KeggRdfFormat]],
          Field(description="aaseq, ntseq, mol, formula, exact_mass, mol_weight, etc."),
      ] = None,
      entries: Annotated[
          Optional[List[str]], Field(description="KEGG entry IDs (e.g., ['hsa:7157', 'hsa00010'])")
      ] = None,
  ) -> str | dict:
      """Execute flexible KEGG API queries across pathways, genes, compounds, diseases, drugs. Use get_kegg_id_by_gene_symbol() first.
  
      Returns:
          str or dict: Raw text response from KEGG API with requested data (pathways, genes, compounds, etc.) or error dict.
      """
      config = KeggConfig(
          operation=operation,
          database=database,
          target_db=target_db,
          source_db=source_db,
          query=query,
          option=option,
          entries=entries or [],
      )
      try:
          KeggConfig.model_validate(config)
      except ValueError as e:
          return {"error": f"Invalid configuration: {e}"}
  
      try:
          return config.execute()
      except Exception as e:
          return {"error": f"Failed to execute KEGG query: {e}"}

• src/biocontext_kb/core/kegg/_query_kegg.py:114-220 (schema)

  KeggConfig Pydantic model with validation and build_path logic that constructs the KEGG API URL path from operation parameters. Contains operation enums (KeggOperation, KeggDatabase, etc.) and the execute() method.

  class KeggConfig(BaseModel):
      """Configuration for KEGG API queries.
  
      This model encapsulates the parameters needed to construct a valid KEGG API request.
      The parameters required depend on the operation being performed.
      """
  
      operation: KeggOperation
      database: Optional[Union[KeggDatabase, KeggOutsideDb, str]] = None
      target_db: Optional[Union[KeggDatabase, KeggOutsideDb, str]] = None
      source_db: Optional[Union[KeggDatabase, KeggOutsideDb, str]] = None
      query: Optional[str] = None
      option: Optional[Union[KeggOption, KeggFindOption, KeggRdfFormat]] = None
      entries: Optional[List[str]] = Field(default_factory=lambda: [])
  
      @field_validator("database", "target_db", "source_db", mode="before")
      @classmethod
      def validate_db(cls, v):
          """Allow organism codes as database values.
  
          This validator handles KEGG organism codes (like 'hsa' for human) as valid database values.
          """
          if v is None:
              return v
          # Check if value is in one of the enums
          try:
              return KeggDatabase(v)
          except ValueError:
              try:
                  return KeggOutsideDb(v)
              except ValueError:
                  # Assume it's an organism code or custom string
                  return v
  
      def build_path(self) -> str:
          """Build the API path based on the configuration.
  
          This method constructs the URL path for the KEGG API request based on the
          operation and parameters provided.
          """
          path_parts = [self.operation.value.lower()]
  
          if self.operation == KeggOperation.INFO:
              if self.database:
                  path_parts.append(str(self.database.value if isinstance(self.database, Enum) else self.database))
  
          elif self.operation == KeggOperation.LIST:
              if self.database:
                  path_parts.append(str(self.database.value if isinstance(self.database, Enum) else self.database))
                  # Special case for pathway/organism
                  if self.database == KeggDatabase.PATHWAY and self.query:
                      path_parts.append(self.query)
                  # Special case for brite/option
                  elif self.database == KeggDatabase.BRITE and self.option:
                      path_parts.append(str(self.option.value if isinstance(self.option, Enum) else self.option))
              elif self.entries:
                  path_parts.append("+".join(self.entries))
  
          elif self.operation == KeggOperation.FIND:
              if self.database and self.query:
                  path_parts.append(str(self.database.value if isinstance(self.database, Enum) else self.database))
                  path_parts.append(self.query)
                  if self.option:
                      path_parts.append(str(self.option.value if isinstance(self.option, Enum) else self.option))
  
          elif self.operation == KeggOperation.GET:
              if self.entries or self.query:
                  if self.entries:
                      path_parts.append("+".join(self.entries))
                  elif self.query:
                      path_parts.append(self.query)
  
                  if self.option:
                      path_parts.append(str(self.option.value if isinstance(self.option, Enum) else self.option))
  
          elif self.operation == KeggOperation.CONV:
              if self.target_db and self.source_db:
                  path_parts.append(str(self.target_db.value if isinstance(self.target_db, Enum) else self.target_db))
                  path_parts.append(str(self.source_db.value if isinstance(self.source_db, Enum) else self.source_db))
              elif self.target_db and self.entries:
                  path_parts.append(str(self.target_db.value if isinstance(self.target_db, Enum) else self.target_db))
                  path_parts.append("+".join(self.entries))
  
          elif self.operation == KeggOperation.LINK:
              if self.target_db and self.source_db:
                  path_parts.append(str(self.target_db.value if isinstance(self.target_db, Enum) else self.target_db))
                  path_parts.append(str(self.source_db.value if isinstance(self.source_db, Enum) else self.source_db))
                  if self.option:
                      path_parts.append(str(self.option.value if isinstance(self.option, Enum) else self.option))
              elif self.target_db and self.entries:
                  path_parts.append(str(self.target_db.value if isinstance(self.target_db, Enum) else self.target_db))
                  path_parts.append("+".join(self.entries))
                  if self.option:
                      path_parts.append(str(self.option.value if isinstance(self.option, Enum) else self.option))
  
          elif self.operation == KeggOperation.DDI and self.entries:
              path_parts.append("+".join(self.entries))
  
          return "/".join(path_parts)
  
      def execute(self) -> str:
          """Execute the API query based on the configuration.
  
          Performs the actual HTTP request to the KEGG API.
          """
          path = self.build_path()
          return execute_kegg_query(path)

• src/biocontext_kb/core/kegg/_query_kegg.py:223-223 (registration)

  Tool registration via the @core_mcp.tool() decorator on line 223, registering the function as an MCP tool named 'query_kegg'.

  @core_mcp.tool()

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

  Exports 'query_kegg' from the kegg package's __init__.py, making it available for import.

  from ._get_kegg_id_by_gene_symbol import get_kegg_id_by_gene_symbol
  from ._query_kegg import query_kegg
  
  __all__ = [
      "get_kegg_id_by_gene_symbol",
      "query_kegg",
  ]

• src/biocontext_kb/core/kegg/_execute_kegg_query.py:1-10 (helper)

  Helper function execute_kegg_query() that performs the actual HTTP GET request to the KEGG REST API at https://rest.kegg.jp and returns raw text.

  import requests
  
  
  def execute_kegg_query(path: str) -> str:
      """Internal helper - executes the HTTP GET and returns raw text."""
      base = "https://rest.kegg.jp"
      url = f"{base}/{path.lstrip('/')}"
      r = requests.get(url, timeout=30.0)
      r.raise_for_status()
      return r.text