query
Execute read-only SELECT queries to explore, debug, or validate data in the Avanti Fellows PostgreSQL database.
Instructions
Execute a read-only SQL query against the database.
Only SELECT queries are allowed. Use this to explore data,
debug issues, or validate assumptions about the data.
Args:
sql: A SELECT query to execute
Returns:
JSON array of results, or error messageInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- mcp_postgres/server.py:47-73 (handler)The primary handler function for the 'query' MCP tool. It validates the SQL as read-only, executes it using asyncpg, serializes results to JSON, and handles errors.
@mcp.tool() async def query(sql: str) -> str: """Execute a read-only SQL query against the database. Only SELECT queries are allowed. Use this to explore data, debug issues, or validate assumptions about the data. Args: sql: A SELECT query to execute Returns: JSON array of results, or error message """ if not is_read_only(sql): return json.dumps({"error": "Only SELECT queries are allowed"}) try: async with get_connection() as conn: rows = await conn.fetch(sql) # Convert to list of dicts, handling special types results = [] for row in rows: results.append({k: _serialize_value(v) for k, v in dict(row).items()}) return json.dumps(results, indent=2, default=str) except Exception as e: return json.dumps({"error": str(e)}) - mcp_postgres/server.py:36-45 (helper)Helper function called by the query handler to ensure only safe, read-only SELECT queries are executed.
def is_read_only(sql: str) -> bool: """Check if SQL is read-only (SELECT only).""" normalized = sql.strip().upper() # Must start with SELECT or WITH (for CTEs) if not (normalized.startswith("SELECT") or normalized.startswith("WITH")): return False # Block dangerous keywords even in subqueries dangerous = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE", "CREATE", "GRANT", "REVOKE"] return not any(kw in normalized for kw in dangerous) - mcp_postgres/server.py:26-34 (helper)Async context manager for obtaining PostgreSQL connections, used by the query handler.
@asynccontextmanager async def get_connection(): """Get a database connection.""" conn = await asyncpg.connect(**DB_CONFIG) try: yield conn finally: await conn.close() - mcp_postgres/server.py:266-276 (helper)Utility function used in query handler to serialize PostgreSQL values (e.g., dates, bytes) to JSON-compatible formats.
def _serialize_value(value): """Serialize special PostgreSQL types to JSON-compatible values.""" if value is None: return None if isinstance(value, (dict, list)): return value if hasattr(value, "isoformat"): # datetime, date, time return value.isoformat() if isinstance(value, bytes): return value.decode("utf-8", errors="replace") return value - mcp_postgres/server.py:47-47 (registration)The @mcp.tool() decorator registers the query function as an MCP tool.
@mcp.tool()