execute_query
Run SQL queries on Vertica databases directly within the MCP Vertica server. Streamline data retrieval and analysis by executing queries efficiently, with support for connection pooling and schema management.
Instructions
Execute a SQL query and return the results.
Args:
ctx: FastMCP context for progress reporting and logging
query: SQL query to execute
database: Optional database name to execute the query against
Returns:
Query results as a string
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes |
Implementation Reference
- src/mcp_vertica/mcp.py:200-246 (handler)The main handler function for the 'execute_query' tool, decorated with @mcp.tool() which also serves as registration. It executes SQL queries on Vertica, handles permissions, fetches results, and returns them as a string.@mcp.tool() async def execute_query(ctx: Context, query: str) -> str: """Execute a SQL query and return the results. Args: ctx: FastMCP context for progress reporting and logging query: SQL query to execute database: Optional database name to execute the query against Returns: Query results as a string """ await ctx.info(f"Executing query: {query}") # Get or create connection manager manager = await get_or_create_manager(ctx) if not manager: return "Error: Failed to initialize database connection. Check configuration." # Extract schema from query if not provided schema = extract_schema_from_query(query) # Check operation permissions operation = extract_operation_type(query) if operation and not manager.is_operation_allowed(schema or "default", operation): error_msg = f"Operation {operation.name} not allowed for schema {schema}" await ctx.error(error_msg) return error_msg conn = None cursor = None try: conn = manager.get_connection() # Always use default DB connection cursor = conn.cursor() cursor.execute(query) results = cursor.fetchall() await ctx.info(f"Query executed successfully, returned {len(results)} rows") return str(results) except Exception as e: error_msg = f"Error executing query: {str(e)}" await ctx.error(error_msg) return error_msg finally: if cursor: cursor.close() if conn: manager.release_connection(conn)