Saltearse al contenido

Python SDK

Esta página aún no está disponible en tu idioma.

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

Source: github.com/pipe2-ai/sdk-python

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="image-generator",
        input={
            "prompt": "A cinematic sunset over mountains, wide angle, golden hour",
            "aspect_ratio": "16:9",
            "quality": "fast",
        },
    )
    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. Register (one call — the server runs the Kratos flow)
signup = await auth.register(
    email="user@example.com",
    password="securepassword",
    name="Jane Doe",
)

# 2. 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,
)

# 3. Create an authenticated client with the token
#    (register returns a null token when email verification is required —
#    sign in after verifying to obtain one)
client = Pipe2Client(signup.register.token)

from pipe2 import Pipe2Client

auth = Pipe2Client("")

# Log in (one call)
login = await auth.login(email="user@example.com", password="securepassword")

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

Available operations

Pipelines

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

# Run a pipeline that consumes an uploaded asset. The input references the
# asset by id — create_asset returns that id after you PUT the file to the
# presigned URL (see the "Assets" section below for the full upload flow).
run = await client.run_pipeline(
    pipeline_slug="transcription",
    input={"source_asset_id": asset_id, "language_code": "auto"},
)

# 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 — use asset.create_asset.id as source_asset_id in
# subsequent pipeline runs.
asset = await client.create_asset(
    asset_type="video",
    url=upload.request_upload.asset_url,
    tags=["input"],
)
asset_id = asset.create_asset.id

# 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 long-running pipeline
run = await client.run_pipeline(
    pipeline_slug="video-generator",
    input={"prompt": "A time-lapse of clouds over a mountain range"},
)

# 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