Postgres MCP

Instructions

Implementation Reference

• src/postgres_mcp/server.py:83-106 (handler)

  The primary handler implementation for the 'list_schemas' MCP tool. It executes a SQL query against information_schema.schemata to retrieve and categorize schemas (system/user), formats the result, and handles errors. The @mcp.tool decorator also serves as the registration.

  @mcp.tool(description="List all schemas in the database")
  async def list_schemas() -> ResponseType:
      """List all schemas in the database."""
      try:
          sql_driver = await get_sql_driver()
          rows = await sql_driver.execute_query(
              """
              SELECT
                  schema_name,
                  schema_owner,
                  CASE
                      WHEN schema_name LIKE 'pg_%' THEN 'System Schema'
                      WHEN schema_name = 'information_schema' THEN 'System Information Schema'
                      ELSE 'User Schema'
                  END as schema_type
              FROM information_schema.schemata
              ORDER BY schema_type, schema_name
              """
          )
          schemas = [row.cells for row in rows] if rows else []
          return format_text_response(schemas)
      except Exception as e:
          logger.error(f"Error listing schemas: {e}")
          return format_error_response(str(e))