SQL Server MCP

Instructions

Implementation Reference

• src/mssql_mcp/tools/export.py:81-162 (handler)

  The complete implementation of the 'export_to_csv' tool handler. The @mcp.tool() decorator handles registration and schema generation from the function signature and docstring. This function validates the SQL query, executes it, and exports the results to a CSV file using the csv module.
  @mcp.tool() def export_to_csv( query: str, filename: str, delimiter: str = ",", ) -> dict[str, Any]: """Export query results to a CSV file. Args: query: SQL SELECT query to execute filename: Output filename (relative or absolute path) delimiter: Field delimiter (default: comma) Returns: Dictionary with: - status: 'success' or error - path: Absolute path to created file - row_count: Number of rows exported - file_size: Size of created file in bytes """ try: manager = get_connection_manager() config = manager.config # Create validator validator = SQLValidator( blocked_commands=config.blocked_commands, read_only=True, 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 for export", "query": query, } # Check blocked commands is_valid, error = validator.validate(query) if not is_valid: return {"error": error, "query": query} # Execute query (no row limit for export) rows = manager.execute_query(query) # Prepare output path path = Path(filename) if not path.is_absolute(): path = Path.cwd() / path # Ensure parent directory exists path.parent.mkdir(parents=True, exist_ok=True) # Handle empty results if not rows: # Create empty file with just a newline path.write_text("") return { "status": "success", "path": str(path.absolute()), "row_count": 0, "file_size": 0, } # Write CSV file with open(path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=rows[0].keys(), delimiter=delimiter) writer.writeheader() writer.writerows(rows) return { "status": "success", "path": str(path.absolute()), "row_count": len(rows), "file_size": path.stat().st_size, } except Exception as e: logger.error(f"Error exporting to CSV: {e}") return {"error": str(e)}