# Python SDK The RailProtocol Python SDK provides a typed interface for making agent payments, managing wallets, setting spend policies, and querying on-chain transaction history. It supports both sync and async usage. *** ## Installation ```bash pip install railprotocol # or uv add railprotocol ``` Requires Python 3.10 or later. *** ## Initialization ```python from railprotocol import RailProtocol import os rail = RailProtocol( agent_id="my-agent", api_key=os.environ["RAIL_API_KEY"], ) ``` **Options:** | Option | Type | Default | Description | | --------------------- | ----- | ------------------------------ | -------------------------------------- | | `agent_id` | str | — | The agent ID. Required. | | `api_key` | str | — | Your RailProtocol API key. Required. | | `base_url` | str | `https://api.railprotocol.org` | Override for self-hosted environments. | | `timeout` | float | `30.0` | Request timeout in seconds. | | `max_retries` | int | `3` | Retries on transient errors. | | `protocol_preference` | str | `"auto"` | `"auto"`, `"x402"`, or `"mpp"`. | *** ## `rail.pay()` Make a payment to an x402 or MPP endpoint. ```python result = rail.pay( endpoint="https://api.dune.com/v1/query/1234/results", method="GET", headers={"X-Custom": "value"}, # optional body={"key": "value"}, # optional, for POST/PUT idempotency_key="run-abc-step-1", # optional ) print(result.protocol) # "x402" print(result.amount_paid) # AmountPaid(usdc=0.002) print(result.tx_hash) # "5yJ8kXtg2xp..." print(result.data) # {"rows": [...], "metadata": {...}} ``` **`PaymentResult` fields:** | Field | Type | Description | | ----------------- | ------------ | ----------------------------- | | `status` | int | HTTP status from the paid API | | `protocol` | str | `"x402"` or `"mpp"` | | `amount_paid` | `AmountPaid` | `AmountPaid(usdc=float)` | | `tx_hash` | str | Solana transaction hash | | `solana_explorer` | str | Link to Solana Explorer | | `data` | Any | Parsed response body | | `headers` | dict | Response headers | | `session_id` | str or None | MPP session ID if used | *** ## `rail.wallet` ### `rail.wallet.get()` ```python wallet = rail.wallet.get() print(wallet.address) # "7xKXtg2xpAGMq8KLwpSodE9LFKq8Z1EscezSShpump" print(wallet.balance.usdc) # 42.50 ``` ### `rail.wallet.balance()` ```python balance = rail.wallet.balance() print(f"Balance: {balance.usdc} USDC") ``` ### `rail.wallet.fund()` ```python rail.wallet.fund(usdc=100) ``` ### `rail.wallet.withdraw()` ```python rail.wallet.withdraw( to="9zXKtg2xpAGMq8KLwpSodE9LFKq8Z1EscezSShpump", usdc=20, ) ``` *** ## `rail.policies` ### `rail.policies.set()` ```python rail.policies.set( daily_budget={"usdc": 100}, per_call_limit={"usdc": 1.00}, allowed_domains=["api.dune.com", "api.browserbase.com", "fal.ai"], rate_limit={"calls": 500, "window": "1h"}, protocol_preference="auto", ) ``` ### `rail.policies.get()` ```python policy = rail.policies.get() print(policy.daily_budget.usdc) # 100.0 print(policy.used_today.usdc) # 12.40 print(policy.per_call_limit.usdc) # 1.0 print(policy.allowed_domains) # ["api.dune.com", ...] ``` *** ## `rail.transactions` ### `rail.transactions.list()` ```python txs = rail.transactions.list( limit=50, protocol="x402", domain="api.dune.com", since="24h", ) for tx in txs: print(tx.tx_hash, tx.amount_paid.usdc, tx.protocol) ``` ### `rail.transactions.get()` ```python tx = rail.transactions.get("5yJ8kXtg2xp...") print(tx.agent_id) print(tx.protocol) print(tx.amount_paid.usdc) print(tx.solana_explorer) ``` ### `rail.transactions.get_chain()` ```python chain = rail.transactions.get_chain(root_tx_hash) for tx in chain: print(f"{tx.agent_id} paid {tx.amount_paid.usdc} USDC (depth {tx.depth})") ``` *** ## Async usage The SDK includes a full async client: ```python import asyncio from railprotocol import AsyncRailProtocol import os rail = AsyncRailProtocol( agent_id="my-agent", api_key=os.environ["RAIL_API_KEY"], ) async def main(): result = await rail.pay( endpoint="https://api.dune.com/v1/query/1234/results", method="GET", ) print(result.amount_paid.usdc) print(result.tx_hash) asyncio.run(main()) ``` ### Parallel payments ```python import asyncio async def parallel_research(): results = await asyncio.gather( rail.pay(endpoint="https://api.dune.com/v1/query/1234/results", method="GET"), rail.pay(endpoint="https://api.coingecko.com/v3/coins/solana", method="GET"), rail.pay(endpoint="https://api.dune.com/v1/query/5678/results", method="GET"), ) total_cost = sum(r.amount_paid.usdc for r in results) print(f"Total: {total_cost} USDC across {len(results)} calls") asyncio.run(parallel_research()) ``` *** ## Error handling ```python from railprotocol import ( RailProtocolError, PolicyViolationError, InsufficientFundsError, DomainNotAllowedError, PaymentProtocolError, PaymentError, AuthenticationError, ) try: result = rail.pay( endpoint="https://api.service.com/data", method="GET", ) except PolicyViolationError as e: print(f"Policy blocked payment: {e.reason}") # "per_call_limit_exceeded: requested 2.50 USDC, limit is 1.00 USDC" except InsufficientFundsError as e: print(f"Insufficient funds. Balance: {e.current_balance.usdc} USDC") except DomainNotAllowedError as e: print(f"Domain not allowed: {e.domain}") except PaymentProtocolError as e: print(f"Unknown protocol at: {e.endpoint}") except PaymentError as e: print(f"Payment failed: {e.message}") except AuthenticationError: print("Invalid API key") ``` *** ## Full example: FastAPI payment endpoint ```python # main.py from fastapi import FastAPI from fastapi.responses import JSONResponse from pydantic import BaseModel from railprotocol import AsyncRailProtocol, PolicyViolationError import os app = FastAPI() rail = AsyncRailProtocol( agent_id=os.environ["AGENT_ID"], api_key=os.environ["RAIL_API_KEY"], ) class PayRequest(BaseModel): endpoint: str method: str = "GET" @app.post("/pay") async def pay(req: PayRequest): try: result = await rail.pay( endpoint=req.endpoint, method=req.method, ) return { "data": result.data, "cost": result.amount_paid.usdc, "tx_hash": result.tx_hash, "protocol": result.protocol, } except PolicyViolationError as e: return JSONResponse( status_code=402, content={"error": "policy_violation", "reason": e.reason}, ) ``` *** ## Full example: autonomous data pipeline ```python import asyncio from railprotocol import AsyncRailProtocol, PolicyViolationError import os rail = AsyncRailProtocol( agent_id="pipeline-agent", api_key=os.environ["RAIL_API_KEY"], ) async def run_pipeline(): # Set policies before running await rail.policies.set( daily_budget={"usdc": 10}, per_call_limit={"usdc": 0.50}, allowed_domains=["api.dune.com", "api.coingecko.com"], ) # Fetch data from multiple sources in parallel try: results = await asyncio.gather( rail.pay(endpoint="https://api.dune.com/v1/query/1234/results", method="GET"), rail.pay(endpoint="https://api.coingecko.com/v3/coins/solana", method="GET"), ) except PolicyViolationError as e: print(f"Aborted: {e.reason}") return total = sum(r.amount_paid.usdc for r in results) print(f"Pipeline complete. Total cost: {total} USDC") print(f"Transactions: {[r.tx_hash for r in results]}") asyncio.run(run_pipeline()) ``` --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://docs.railprotocol.org/sdk/python.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.