M4

Instructions

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",
      }
  )