Postgres MCP Pro

Instructions

Implementation Reference

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

  The list_schemas tool is defined as an async function decorated with @mcp.tool. It retrieves schemas by querying information_schema.schemata using an SQL driver.

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