execute_sql
Execute SQL queries to manage PostgreSQL databases, enabling schema operations, query optimization, performance monitoring, and health analysis.
Instructions
Execute any SQL query
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | No | SQL to run | all |
Implementation Reference
- The core handler function for the execute_sql tool. It takes a SQL string, executes it via sql_driver.execute_query, formats the results, and handles errors.
async def execute_sql( sql: str = Field(description="SQL to run", default="all"), ) -> ResponseType: """Executes a SQL query against the database.""" try: sql_driver = await get_sql_driver() rows = await sql_driver.execute_query(sql) # type: ignore if rows is None: return format_text_response("No results") results = [r.cells for r in rows] return format_text_response(format_query_results_as_text(results)) except Exception as e: logger.error(f"Error executing query: {e}") return format_error_response(str(e)) - src/postgres_mcp_pro_plus/server.py:784-787 (registration)Registers the execute_sql tool with the MCP server, using different descriptions based on the access mode (unrestricted or restricted/read-only).
if current_access_mode == AccessMode.UNRESTRICTED: mcp.add_tool(execute_sql, description="Execute any SQL query") else: mcp.add_tool(execute_sql, description="Execute a read-only SQL query") - Pydantic Field definition for the sql input parameter, defining the input schema for the tool.
sql: str = Field(description="SQL to run", default="all"),