BioContextAI Knowledgebase MCP

Instructions

Implementation Reference

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

  Main handler function for the 'bc_query_kegg' tool. Decorated with @core_mcp.tool(), constructs KeggConfig from inputs, validates, 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-221 (schema)

  Pydantic model KeggConfig and its methods for validating inputs, building API paths, and executing queries. Includes field validators for databases. Used by the handler for schema validation.
  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/app.py:35-39 (registration)

  Registers the core_mcp server (named 'BC', slugged to 'bc') into the main FastMCP app, prefixing its tools (e.g., 'query_kegg' becomes 'bc_query_kegg').
  for mcp in [core_mcp, *(await get_openapi_mcps())]: await mcp_app.import_server( mcp, slugify(mcp.name), )
• src/biocontext_kb/core/kegg/_execute_kegg_query.py:1-11 (helper)

  Helper function that performs the actual HTTP GET request to the KEGG REST API endpoint.
  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
• src/biocontext_kb/core/kegg/__init__.py:1-7 (registration)

  Imports the query_kegg function, triggering its registration via the @core_mcp.tool() decorator when the module is imported.
  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/_server.py:1-6 (registration)

  Defines the core_mcp FastMCP server instance named 'BC', to which tools like query_kegg are attached via decorators.
  from fastmcp import FastMCP core_mcp = FastMCP( # type: ignore "BC", instructions="Provides access to biomedical knowledge bases.", )