execute_query
Execute read-only SELECT queries on SQL Server databases to retrieve data with automatic row limiting for safety and performance.
Instructions
Execute a read-only SQL query and return results.
Only SELECT statements are allowed. The query will have a row limit applied
automatically if not specified.
Args:
query: SQL SELECT statement to execute
max_rows: Maximum rows to return (overrides default, capped by MSSQL_MAX_ROWS)
Returns:
Dictionary with:
- query: The original query
- executed_query: The query that was actually executed (may include TOP)
- columns: List of column names
- rows: List of row dictionaries
- row_count: Number of rows returned
- max_rows: The effective row limit appliedInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| max_rows | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/mssql_mcp/tools/query.py:14-82 (handler)Core handler function for the 'execute_query' MCP tool. Includes registration via @mcp.tool() decorator, input schema via type hints and docstring, validation logic, query execution, and result formatting.
def execute_query(query: str, max_rows: int | None = None) -> dict[str, Any]: """Execute a read-only SQL query and return results. Only SELECT statements are allowed. The query will have a row limit applied automatically if not specified. Args: query: SQL SELECT statement to execute max_rows: Maximum rows to return (overrides default, capped by MSSQL_MAX_ROWS) Returns: Dictionary with: - query: The original query - executed_query: The query that was actually executed (may include TOP) - columns: List of column names - rows: List of row dictionaries - row_count: Number of rows returned - max_rows: The effective row limit applied """ try: manager = get_connection_manager() config = manager.config # Create validator validator = SQLValidator( blocked_commands=config.blocked_commands, read_only=True, # execute_query is always read-only allowed_schemas=config.allowed_schemas if config.allowed_schemas else None, ) # Validate query is SELECT-only if not validator.is_select_only(query): return { "error": "Only SELECT queries are allowed. Use other tools for data modification.", "query": query, } # Check blocked commands is_valid, error = validator.validate(query) if not is_valid: return {"error": error, "query": query} # Determine effective row limit effective_max_rows = min(max_rows or config.max_rows, config.max_rows) # Inject row limit executed_query = validator.inject_row_limit(query, effective_max_rows) # Execute query rows = manager.execute_query(executed_query) # Extract column names from first row or return empty columns: list[str] = [] if rows: columns = list(rows[0].keys()) return { "query": query, "executed_query": executed_query, "columns": columns, "rows": rows, "row_count": len(rows), "max_rows": effective_max_rows, } except Exception as e: logger.error(f"Error executing query: {e}") return {"error": str(e), "query": query} - src/mssql_mcp/tools/query.py:15-32 (schema)Input/output schema defined by function signature type hints and comprehensive docstring describing parameters and return structure.
"""Execute a read-only SQL query and return results. Only SELECT statements are allowed. The query will have a row limit applied automatically if not specified. Args: query: SQL SELECT statement to execute max_rows: Maximum rows to return (overrides default, capped by MSSQL_MAX_ROWS) Returns: Dictionary with: - query: The original query - executed_query: The query that was actually executed (may include TOP) - columns: List of column names - rows: List of row dictionaries - row_count: Number of rows returned - max_rows: The effective row limit applied """ - src/mssql_mcp/tools/query.py:14-14 (registration)Tool registration using the @mcp.tool() decorator from FastMCP.
def execute_query(query: str, max_rows: int | None = None) -> dict[str, Any]: