Skip to content

README.md

Latest commit

 

History

History
177 lines (134 loc) · 5.65 KB

README.md

File metadata and controls

177 lines (134 loc) · 5.65 KB
Raw

Pipe Python SDK

pipe-sdk is the synchronous Python client for Pipe's native object API, Durable Objects HTTP API, storage-credit billing API, and S3-compatible endpoint. It uses httpx; the optional S3 helper configures boto3 with the settings Pipe requires.

Durable Objects are operator-only today. Production Durable Object calls require the router's operator write token, including reads. Do not ship that token to browsers or untrusted applications.

S3 credentials are provisioned externally today. Create them in the authenticated storage workspace at https://pipe.love/storage. This SDK consumes an access key and one-time secret; it cannot create them.

The package exports DEFAULT_ONBOARDING_URL for that storage workspace URL.

Install

python -m pip install 'pipe-sdk[s3]'

From the repository root, use python -m pip install -e 'python[s3]' instead.

Fastest public onboarding

Buy credit and create a scoped S3 credential at pipe.love/storage, then export the one-time credential and use the S3 helper:

export PIPE_S3_ACCESS_KEY_ID=LT...
export PIPE_S3_SECRET_ACCESS_KEY=shown-once
from pipe_sdk import create_s3_client

s3 = create_s3_client()
s3.put_object(Bucket="example", Key="hello.txt", Body=b"hello")
body = s3.get_object(Bucket="example", Key="hello.txt")["Body"].read()

Configuration

Endpoints default to the public beta. Storage operations still require the credential appropriate to the selected API. Every value can also be passed explicitly to PipeConfig.

export PIPE_STORAGE_ENDPOINT=https://api.pipedev.network
export PIPE_BILLING_ENDPOINT=https://a.pipenetwork.ai/control-api
export PIPE_DATABASE=default

# Only for a trusted/self-hosted router or operator environment:
export PIPE_TOKEN=replace-me

The aliases PIPE_ROUTER_URL, PIPE_CONTROL_PLANE_URL, and PIPE_BEARER_TOKEN are also accepted. PIPE_TIMEOUT_SECONDS changes the default 60-second HTTP timeout.

Native storage (operator or external voucher)

from pipe_sdk import PipeClient

with PipeClient.from_env() as pipe:
    stored = pipe.storage.put("hello.txt", b"hello")
    print(stored.etag)
    print(pipe.storage.get("hello.txt"))
    print(pipe.storage.head("hello.txt").bytes)
    pipe.storage.delete("hello.txt")

The snippet above requires an operator bearer token or a self-hosted router that does not require payment. The public paid router requires an externally constructed cumulative voucher, supplied as shown below.

On a paid router, native operations use Pipe's cumulative storage voucher. Construct that voucher outside this package and pass it per operation:

pipe.storage.put("hello.txt", b"hello", payment_signature=encoded_voucher)

payment_signature here is not the same payload as the exact Solana USDC payment used to buy credit.

Native get and head explicitly use the router's proxy mode so credentials and short-lived grants are never forwarded through a redirect. Whole objects above the router's 32 MiB proxy ceiling need S3 range reads or an explicit direct-read grant workflow.

The native router accepts payment headers up to 16 KiB. The SDK validates that limit and header-safe encoding before making a request.

Buy storage credit

The SDK owns the invoice/retry/polling flow but deliberately does not hold a wallet key or choose a Solana signer. Supply a callback that receives the exact payment challenge and returns the base64 x402 PAYMENT-SIGNATURE value.

from pipe_sdk import PipeClient, TopUpRequest

def sign_exact_usdc(challenge: TopUpRequest) -> str:
    # Validate challenge.payment_required, construct and sign the exact SVM
    # transaction with your wallet integration, then return the encoded header.
    return wallet_adapter.sign_x402(challenge.payment_required)

with PipeClient.from_env() as pipe:
    invoice = pipe.billing.top_up(
        wallet="your-solana-wallet",
        amount_atoms=1_000_000,
        signer=sign_exact_usdc,
    )
    print(invoice.state, invoice.transaction)

Use billing.pricing() for the public read/write price schedule. For manual payment control use billing.config(), billing.create(), billing.submit(), and billing.poll(). create_top_up, submit_top_up, and poll_top_up are descriptive aliases.

S3

Install the s3 extra and provide credentials created at pipe.love:

export PIPE_S3_ACCESS_KEY_ID=LT...
export PIPE_S3_SECRET_ACCESS_KEY=shown-once
from pipe_sdk import create_s3_client

s3 = create_s3_client()
s3.put_object(Bucket="example", Key="hello.txt", Body=b"hello")
body = s3.get_object(Bucket="example", Key="hello.txt")["Body"].read()

The factory sets path-style addressing, SigV4, and checksum-on-demand behavior.

Durable Objects

from pipe_sdk import PipeClient, SqlStatement

with PipeClient.from_env() as pipe:
    durable = pipe.durable_objects
    durable.create_namespace("chat")
    object_id = durable.issue_id("chat", name="room:general").id
    durable.open_object("chat", object_id)
    durable.put_state("chat", object_id, "profile", {"members": 3})
    print(durable.get_state("chat", object_id, "profile").value)
    results = durable.transaction(
        "chat",
        object_id,
        [SqlStatement("SELECT 1 AS value")],
    )
    print(results[0].rows)

The client covers namespace and object lifecycle, stable IDs, JSON state, SQL, transactions, migrations, and object-scoped blobs. Retrying a non-idempotent SQL mutation after an ambiguous network failure can apply it twice; encode an operation ID and uniqueness constraint in your schema.

Tests

python -m pip install -e '.[test]'
pytest