Avanti Fellows PostgreSQL MCP Server

Instructions

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()