export_to_json
Execute SQL SELECT queries and export the results directly to JSON files for data analysis, sharing, or integration with other applications.
Instructions
Export query results to a JSON file.
Args:
query: SQL SELECT query to execute
filename: Output filename (relative or absolute path)
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 bytesInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| filename | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/mssql_mcp/tools/export.py:16-78 (handler)The export_to_json tool handler, including decorator for MCP registration, input validation, SQL execution, and JSON file export logic.
@mcp.tool() def export_to_json(query: str, filename: str) -> dict[str, Any]: """Export query results to a JSON file. Args: query: SQL SELECT query to execute filename: Output filename (relative or absolute path) 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) # Write JSON file with open(path, "w", encoding="utf-8") as f: json.dump(rows, f, indent=2, default=str) 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 JSON: {e}") return {"error": str(e)}