execute_sql
Execute custom SQL queries to analyze bakery sales data, transaction records, and product catalogs from the POS database for business insights.
Instructions
Execute a custom SQL query on the database. Use with caution. Read-only queries recommended.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | SQL query to execute. | |
| params | No | Parameters for the SQL query (for parameterized queries). Optional. |
Implementation Reference
- src/bakery_data_mcp/server.py:204-224 (registration)Registers the 'execute_sql' tool with MCP Server, including name, description, and input schema for query and optional params.Tool( name="execute_sql", description="Execute a custom SQL query on the database. Use with caution. Read-only queries recommended.", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "SQL query to execute." }, "params": { "type": "array", "description": "Parameters for the SQL query (for parameterized queries). Optional.", "items": { "type": ["string", "number", "null"] } } }, "required": ["query"] } ),
- src/bakery_data_mcp/server.py:517-535 (handler)Handler function for 'execute_sql' tool: executes the SQL query with parameters using the database cursor, returns JSON results for SELECT queries or success message with row count for others.elif name == "execute_sql": query = arguments["query"] params = arguments.get("params", []) cursor.execute(query, params) # Check if this is a SELECT query if query.strip().upper().startswith("SELECT"): results = cursor.fetchall() return [TextContent( type="text", text=json.dumps(results, ensure_ascii=False, indent=2) )] else: conn.commit() return [TextContent( type="text", text=f"Query executed successfully. Rows affected: {cursor.rowcount}" )]