opengenes_db_query

Instructions

Implementation Reference

• src/opengenes_mcp/server.py:250-289 (handler)

  The handler function for the opengenes_db_query tool. Executes the provided SQL query using the DatabaseManager and returns the QueryResult.

  def db_query(self, sql: str) -> QueryResult:
      """
      Execute a read-only SQL query against the OpenGenes database.
      
      SECURITY: This tool uses SQLite's built-in read-only mode to enforce data protection.
      The database connection is opened in read-only mode, preventing any write operations
      at the database level. Any attempt to modify data will be automatically rejected by SQLite.
      
      The OpenGenes database contains aging and lifespan research data across 4 tables:
      - lifespan_change: Experimental data on gene modifications and lifespan effects
      - gene_criteria: Aging-related criteria classifications for genes  
      - gene_hallmarks: Links genes to hallmarks of aging (multi-value field)
      - longevity_associations: Population genetics data on gene variants and longevity
      
      All tables are linked by HGNC gene symbols for cross-table analysis.
      
      CRITICAL REMINDERS:
      - Use LIKE queries with wildcards for multi-value fields (gene_hallmarks."hallmarks of aging", 
        lifespan_change.intervention_improves, lifespan_change.intervention_deteriorates)
      - Order lifespan results by magnitude: DESC for increases, ASC for decreases
      - IMPORTANT: When user asks about "lifespan effects" without specifying mean vs max, 
        show both lifespan_percent_change_mean AND lifespan_percent_change_max
      - Use COALESCE(lifespan_percent_change_mean, lifespan_percent_change_max) for ordering both metrics
      - COMPREHENSIVE AGING EVIDENCE: When users ask about "evidence of aging", "link to aging/longevity", 
        or "aging associations" for a gene, query ALL 4 tables (gene_criteria, gene_hallmarks, 
        lifespan_change, longevity_associations) for complete scientific evidence
      
      For detailed schema information, use get_schema_info().
      For query examples and patterns, use get_example_queries().
      
      Args:
          sql: The SQL query to execute (database enforces read-only access)
          
      Returns:
          QueryResult: Contains the query results, row count, and executed query
      """
      with start_action(action_type="db_query_tool", sql=sql) as action:
          result = self.db_manager.execute_query(sql)
          return result

• src/opengenes_mcp/server.py:166-180 (registration)

  Registers the opengenes_db_query tool using FastMCP's tool decorator with name constructed from prefix + 'db_query' and binds it to the db_query handler.

  def _register_opengenes_tools(self):
      """Register OpenGenes-specific tools."""
      self.tool(name=f"{self.prefix}get_schema_info", description="Get information about the database schema")(self.get_schema_info)
      self.tool(name=f"{self.prefix}example_queries", description="Get a list of example SQL queries")(self.get_example_queries)
      description = "Query the Opengenes database that contains data about genes involved in longevity, lifespan extension experiments on model organisms, and changes in human and other organisms with aging."
      if self.huge_query_tool:
          # Load and concatenate the prompt from package data
          prompt_content = get_prompt_content().strip()
          if prompt_content:
              description = description + "\n\n" + prompt_content
          self.tool(name=f"{self.prefix}db_query", description=description)(self.db_query)
      else:
          description = description + " Before caling this tool the first time, always check tools that provide schema information and example queries."
          self.tool(name=f"{self.prefix}db_query", description=description)(self.db_query)

• src/opengenes_mcp/server.py:90-95 (schema)

  Pydantic model defining the output schema for the db_query tool response.

  class QueryResult(BaseModel):
      """Result from a database query."""
      rows: List[Dict[str, Any]] = Field(description="Query result rows")
      count: int = Field(description="Number of rows returned")
      query: str = Field(description="The SQL query that was executed")

• src/opengenes_mcp/server.py:104-141 (helper)

  DatabaseManager.execute_query: Core helper that performs the actual SQL execution in read-only mode, used by the db_query handler.

  def execute_query(self, sql: str, params: Optional[tuple] = None) -> QueryResult:
      """Execute a read-only SQL query and return results."""
      with start_action(action_type="execute_query", sql=sql, params=params) as action:
          # Execute query using read-only connection - SQLite will enforce read-only at database level
          # Using URI format with mode=ro for true read-only access
          readonly_uri = f"file:{self.db_path}?mode=ro"
          
          try:
              with sqlite3.connect(readonly_uri, uri=True) as conn:
                  conn.row_factory = sqlite3.Row  # This allows dict-like access to rows
                  cursor = conn.cursor()
                  
                  if params:
                      cursor.execute(sql, params)
                  else:
                      cursor.execute(sql)
                  
                  rows = cursor.fetchall()
                  # Convert sqlite3.Row objects to dictionaries
                  rows_dicts = [dict(row) for row in rows]
                  
                  result = QueryResult(
                      rows=rows_dicts,
                      count=len(rows_dicts),
                      query=sql
                  )
                  
                  action.add_success_fields(rows_count=len(rows_dicts))
                  return result
          except sqlite3.OperationalError as e:
              if "readonly database" in str(e).lower():
                  error_msg = f"Write operation attempted on read-only database. Rejected query: {sql}"
                  action.log(message_type="query_rejected", error=error_msg, rejected_query=sql)
                  raise ValueError(error_msg) from e
              else:
                  # Re-raise other operational errors
                  raise

• src/opengenes_mcp/server.py:145-165 (registration)

  OpenGenesMCP.__init__: Initializes the server, sets the tool prefix to 'opengenes_', and calls _register_opengenes_tools() to register the tools.

  def __init__(
      self, 
      name: str = "OpenGenes MCP Server",
      db_path: Path = DB_PATH,
      prefix: str = "opengenes_",
      huge_query_tool: bool = False,
      **kwargs
  ):
      """Initialize the OpenGenes tools with a database manager and FastMCP functionality."""
      # Initialize FastMCP with the provided name and any additional kwargs
      super().__init__(name=name, **kwargs)
      
      # Initialize our database manager
      self.db_manager = DatabaseManager(db_path)
      
      self.prefix = prefix
      self.huge_query_tool = huge_query_tool
      # Register our tools and resources
      self._register_opengenes_tools()
      self._register_opengenes_resources()