execute_query
Execute SQL SELECT queries to analyze clinical datasets like MIMIC-IV and eICU, returning query results or helpful error messages for data analysis.
Instructions
🚀 Execute SQL queries to analyze data.
Recommended workflow:
Use get_database_schema() to list tables
Use get_table_info() to examine structure
Write your SQL query with exact names
Args: sql_query: Your SQL SELECT query (SELECT only).
Returns: Query results or helpful error messages.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql_query | Yes |
Implementation Reference
- src/m4/core/tools/tabular.py:180-200 (handler)Core handler implementation for the execute_query tool. Performs safety validation, executes via backend, handles errors.def invoke( self, dataset: DatasetDefinition, params: ExecuteQueryInput ) -> ToolOutput: """Execute a SQL query with safety validation.""" # Validate query first safe, msg = is_safe_query(params.sql_query) if not safe: return ToolOutput(result=f"**Security Error:** {msg}") backend = get_backend() try: result = backend.execute_query(params.sql_query, dataset) if result.success: return ToolOutput(result=result.data) else: return ToolOutput( result=format_error_with_guidance(result.error or "Unknown error") ) except Exception as e: return ToolOutput(result=format_error_with_guidance(str(e)))
- src/m4/core/tools/tabular.py:40-45 (schema)Input schema definition (dataclass) for the execute_query tool.@dataclass class ExecuteQueryInput(ToolInput): """Input for execute_query tool.""" sql_query: str
- src/m4/core/tools/__init__.py:66-66 (registration)Registration of the ExecuteQueryTool instance in the internal ToolRegistry during init_tools().ToolRegistry.register(ExecuteQueryTool())
- src/m4/mcp_server.py:153-178 (handler)Thin MCP adapter/handler for execute_query: performs capability check and delegates to core tool.@mcp.tool() @require_oauth2 def execute_query(sql_query: str) -> str: """🚀 Execute SQL queries to analyze data. **Recommended workflow:** 1. Use get_database_schema() to list tables 2. Use get_table_info() to examine structure 3. Write your SQL query with exact names Args: sql_query: Your SQL SELECT query (SELECT only). Returns: Query results or helpful error messages. """ dataset = DatasetRegistry.get_active() # Proactive capability check result = _tool_selector.check_compatibility("execute_query", dataset) if not result.compatible: return result.error_message tool = ToolRegistry.get("execute_query") return tool.invoke(dataset, ExecuteQueryInput(sql_query=sql_query)).result
- src/m4/mcp_server.py:47-58 (registration)MCP tool names set including 'execute_query' for filtering and snapshots._MCP_TOOL_NAMES = frozenset( { "list_datasets", "set_dataset", "get_database_schema", "get_table_info", "execute_query", "search_notes", "get_note", "list_patient_notes", } )