OpenGenes MCP Server

Instructions

Implementation Reference

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

  The db_query method implements the core logic of the 'opengenes_db_query' tool. It receives the SQL query as input, delegates execution to DatabaseManager.execute_query, 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-181 (registration)

  The _register_opengenes_tools method registers the 'opengenes_db_query' tool using FastMCP's self.tool decorator with name '{prefix}db_query' (prefix='opengenes_' by default) bound 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 BaseModel defining the output schema for the tool response, including query results (rows), row count, and the executed SQL query.
  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)

  The execute_query method in DatabaseManager class performs the actual SQLite database query execution in read-only mode, handles errors, converts rows to dicts, and constructs QueryResult. This is called by the tool 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:162-163 (registration)

  In the OpenGenesMCP __init__ method, _register_opengenes_tools() is called to register all tools including 'opengenes_db_query'.
  # Register our tools and resources self._register_opengenes_tools()