execute_query
Run SQL queries against Vertica databases to retrieve data results. Supports optional database targeting for flexible query execution.
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 stringInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes |
Implementation Reference
- src/mcp_vertica/mcp.py:200-245 (registration)The `execute_query` tool is registered via the `@mcp.tool()` decorator on line 200, which makes it available as an MCP tool.
@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) - src/mcp_vertica/mcp.py:200-245 (handler)The `execute_query` function is the handler that executes a SQL query. It extracts the operation type and schema from the query, checks permissions, executes the query via the Vertica connection pool, fetches all 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) - src/mcp_vertica/mcp.py:83-95 (helper)`extract_operation_type` helper used by `execute_query` to determine if the query is INSERT, UPDATE, DELETE, or DDL for permission checking.
def extract_operation_type(query: str) -> OperationType | None: """Extract the operation type from a SQL query.""" query = query.strip().upper() if query.startswith("INSERT"): return OperationType.INSERT elif query.startswith("UPDATE"): return OperationType.UPDATE elif query.startswith("DELETE"): return OperationType.DELETE elif any(query.startswith(op) for op in ["CREATE", "ALTER", "DROP", "TRUNCATE"]): return OperationType.DDL return None - src/mcp_vertica/mcp.py:98-104 (helper)`extract_schema_from_query` helper used by `execute_query` to parse the schema name from a fully-qualified table reference in the SQL query.
def extract_schema_from_query(query: str) -> str | None: """Extract schema name from a SQL query.""" # database.table 또는 schema.table 패턴에서 schema 추출 match = re.search(r"([a-zA-Z0-9_]+)\.[a-zA-Z0-9_]+", query) if match: return match.group(1) return None - `is_operation_allowed` method on `VerticaConnectionManager` used by `execute_query` to check if a given operation type is permitted for the target schema.
def is_operation_allowed(self, database: str, operation: OperationType) -> bool: """Check if an operation is allowed for a specific database.""" if not self.config: return False # Get schema permissions schema_permissions = self.config.schema_permissions or {} schema_perms = schema_permissions.get(database) # Check schema-specific permissions first if schema_perms: if operation == OperationType.INSERT: return schema_perms.insert elif operation == OperationType.UPDATE: return schema_perms.update elif operation == OperationType.DELETE: return schema_perms.delete elif operation == OperationType.DDL: return schema_perms.ddl # Fall back to global permissions if operation == OperationType.INSERT: return self.config.allow_insert elif operation == OperationType.UPDATE: return self.config.allow_update elif operation == OperationType.DELETE: return self.config.allow_delete elif operation == OperationType.DDL: return self.config.allow_ddl return False