Skip to content

Python SDK

The official Pipe2.ai Python SDK provides fully typed async GraphQL operations with Pydantic models. Generated with ariadne-codegen.

Installation

pip install pipe2

Quick start

import asyncio
from pipe2 import Pipe2Client

async def main():
    client = Pipe2Client("YOUR_JWT_TOKEN")

    # List all pipelines
    result = await client.get_pipelines()
    for p in result.pipelines:
        print(f"{p.name} — {p.credit_cost} credits")

    # Run a pipeline
    run = await client.run_pipeline(
        pipeline_slug="selfie-universe",
        input={"photo_url": "https://..."},
    )
    print(f"Run ID: {run.run_pipeline.run_id}")

asyncio.run(main())

Authentication

Pass your JWT token when creating the client:

client = Pipe2Client("eyJhbGciOiJFZERTQSIs...")

from pipe2 import Pipe2Client

# Use an unauthenticated client for auth flows
auth = Pipe2Client("")

# 1. Start a signup flow
flow = await auth.init_signup_flow()

# 2. Submit signup credentials
signup = await auth.submit_signup_flow(
    flow_id=flow.init_signup_flow.id,
    email="user@example.com",
    password="securepassword",
    name="Jane Doe",
    csrf_token=flow.init_signup_flow.csrf_token,
)

# 3. Verify email (check inbox for the 6-digit code)
vflow = await auth.init_verification_flow(email="user@example.com")
await auth.submit_verification_code(
    flow_id=vflow.init_verification_flow.id,
    code="123456",
    csrf_token=vflow.init_verification_flow.csrf_token,
)

# 4. Create an authenticated client with the token
client = Pipe2Client(signup.submit_signup_flow.token)

from pipe2 import Pipe2Client

auth = Pipe2Client("")

# 1. Start a login flow
flow = await auth.init_login_flow()

# 2. Submit credentials
login = await auth.submit_login_flow(
    flow_id=flow.init_login_flow.id,
    email="user@example.com",
    password="securepassword",
    csrf_token=flow.init_login_flow.csrf_token,
)

# 3. Use the token
client = Pipe2Client(login.submit_login_flow.token)
pipelines = await client.get_pipelines()

social = await auth.submit_social_login(provider="google")  # or "github", "microsoft"
# Redirect user to social.submit_social_login.redirect_url

Available operations

Pipelines

# List all pipelines
result = await client.get_pipelines()

# Run a pipeline
run = await client.run_pipeline(
    pipeline_slug="baby-generator",
    input={"photo_url_1": "https://...", "photo_url_2": "https://..."},
)

# Get a specific run
run_detail = await client.get_pipeline_run(id="run-uuid")

# List run history with pagination
runs = await client.get_pipeline_runs(limit=10, offset=0)

Assets

# Get a presigned upload URL
upload = await client.request_upload(filename="photo.jpg", content_type="image/jpeg")
# PUT the file to upload.request_upload.upload_url

# Create an asset record
asset = await client.create_asset(
    asset_type="image",
    url=upload.request_upload.asset_url,
    tags=["input", "selfie-universe"],
)

# List your assets
assets = await client.get_user_assets(limit=20)

# Delete an asset (also removes from storage)
await client.delete_asset_action(id="asset-uuid")

# Update tags
await client.update_asset_tags(id="asset-uuid", tags=["favorite"])

Credits

# Check your balance
balance = await client.get_credit_balance()
print(f"Available: {balance.get_credit_balance.available}")

# View transaction history
history = await client.get_credit_history()

Personal access tokens

# Create a new token
token = await client.create_personal_access_token(name="My Integration")
print(token.create_personal_access_token.token)  # Save this — shown only once

# List tokens
tokens = await client.get_my_api_keys()

# Revoke a token
await client.revoke_personal_access_token(id="token-uuid")

Plans & billing

plans = await client.get_plans()
credit_packs = await client.get_credit_packs()
subscription = await client.get_subscription()

Subscriptions

Watch a pipeline run status in real time. The SDK returns a typed async iterator:

# Start a pipeline
run = await client.run_pipeline(
    pipeline_slug="selfie-universe",
    input={"photo_url": "https://..."},
)

# Subscribe — async iterator with typed updates
async for result in client.watch_pipeline_run(run_id=run.run_pipeline.run_id):
    r = result.pipeline_runs_by_pk
    print(f"Status: {r.status}")

    if r.status in ("completed", "failed"):
        if r.assets:
            print(f"Result: {r.assets[0].url}")
        break