Python SDK

The subcent Python package provides an async client for the Subcent API, with built-in LangChain integration for AI agent workflows.

Installation#

pip install subcent

For LangChain integration:

pip install subcent[langchain]

Quick Start#

from subcent import Subcent

async def main():
    sc = Subcent(
        api_key="sc_test_your_api_key",
        network="testnet",
        base_url="https://testnet-api.subcent.io",
    )

    try:
        payment = await sc.payments.request({
            "agent_id": "agt_550e8400...",
            "merchant_id": "mrc_660e8400...",
            "amount": 2500,
            "currency": "USDC",
            "category": "groceries",
        })
        print(payment["status"])
    finally:
        await sc.close()

Client Configuration#

sc = Subcent(
    api_key="sc_test_your_api_key",
    base_url="https://testnet-api.subcent.io",  # default: http://localhost:3000
    network="testnet",                            # "devnet", "testnet", or "mainnet"
)

| Parameter | Type | Required | Description | |---|---|---|---| | api_key | str | Yes | Your API key (sc_live_, sc_test_, or sc_agent_ prefix) | | base_url | str | No | API base URL (default: http://localhost:3000) | | network | str | No | Network identifier (default: "devnet") |

The client uses httpx.AsyncClient internally and sets the Authorization and Content-Type headers automatically. All API methods are async and must be awaited.

Always call await sc.close() when you are done to release the underlying HTTP connection pool. Use a try/finally block or an async context manager.

Payments#

payments.request(params)#

Submit a payment request.

payment = await sc.payments.request({
    "agent_id": "agt_550e8400...",
    "merchant_id": "mrc_660e8400...",
    "amount": 2500,
    "currency": "USDC",
    "category": "groceries",
    "memo": "Weekly grocery order",
})

print(payment["status"])  # "completed" | "pending_approval" | "rejected"

PaymentRequest TypedDict:

| Field | Type | Required | Description | |---|---|---|---| | agent_id | str | Yes | The agent making the payment | | merchant_id | str | Yes | The merchant receiving the payment | | amount | int | Yes | Amount in minor units | | currency | str | No | Currency code (default: "USDC") | | category | str | Yes | Spending category | | memo | str | No | Optional payment memo |

payments.get(id)#

Retrieve a specific payment by ID.

payment = await sc.payments.get("pay_770e8400...")
print(payment["status"])
print(payment.get("tx_hash"))

payments.list(**params)#

List payments with optional filters.

result = await sc.payments.list(
    vault_id="550e8400...",
    agent_id="agt_550e8400...",
    status="completed",
    limit=20,
)

for p in result["payments"]:
    print(p["payment_id"], p["amount"])

payments.approve(id)#

Approve a payment pending human approval.

result = await sc.payments.approve("pay_770e8400...")
print(result["status"])  # "completed"

payments.reject(id, reason=None)#

Reject a payment pending human approval.

result = await sc.payments.reject("pay_770e8400...", reason="Amount too high")
print(result["status"])  # "rejected"

Vaults#

vaults.create(params)#

Create a new vault.

vault = await sc.vaults.create({
    "owner_wallet": "7xKXtg2CW87d97TXJSDpbD5jBkheTqA83TZRuJosgAsU",
    "chain": "solana",
    "label": "Production Vault",
})

print(vault["vault_id"])
print(vault["vault_address"])

vaults.get(id)#

Retrieve vault details.

vault = await sc.vaults.get("550e8400...")
print(vault["balance"])
print(vault["status"])

vaults.freeze(id)#

Freeze a vault.

result = await sc.vaults.freeze("550e8400...")
print(result["status"])  # "frozen"

vaults.unfreeze(id)#

Unfreeze a vault.

result = await sc.vaults.unfreeze("550e8400...")
print(result["status"])  # "active"

Policies#

policies.create(params)#

Create a spending policy.

policy = await sc.policies.create({
    "agent_id": "agt_550e8400...",
    "vault_id": "550e8400...",
    "budget": {
        "max_per_transaction": 100,
        "max_daily": 500,
        "max_monthly": 5000,
        "currency": "USDC",
    },
    "categories": {
        "mode": "allowlist",
        "list": ["groceries", "household"],
    },
    "velocity_controls": {
        "max_transactions_per_hour": 10,
        "max_transactions_per_day": 50,
    },
})

print(policy["policy_id"])
print(policy["policy_hash"])

policies.get(id)#

Retrieve a policy.

policy = await sc.policies.get("pol_880e8400...")
print(policy["budget"]["max_per_transaction"])

policies.update(id, params)#

Update a policy.

updated = await sc.policies.update("pol_880e8400...", {
    "budget": {
        "max_per_transaction": 200,
        "max_daily": 1000,
        "currency": "USDC",
    },
})

print(updated["version"])  # incremented

Escrows#

escrows.create(params)#

Create a conditional escrow.

escrow = await sc.escrows.create({
    "merchant_id": "mrc_660e8400...",
    "amount": 50000,
    "currency": "USDC",
    "condition": {
        "type": "delivery_confirmation",
        "data": {"tracking_number": "1Z999AA10123456784"},
    },
    "expires_in_hours": 72,
})

print(escrow["escrow_id"])
print(escrow["status"])  # "funded"

escrows.get(id)#

Retrieve escrow details.

escrow = await sc.escrows.get("esc_990e8400...")
print(escrow["status"])
print(escrow["expires_at"])

escrows.release(id)#

Release escrow funds to the merchant.

result = await sc.escrows.release("esc_990e8400...")
print(result["status"])  # "released"

escrows.refund(id)#

Refund escrow funds back to the vault.

result = await sc.escrows.refund("esc_990e8400...")
print(result["status"])  # "refunded"

Catalog#

catalog.search(merchant_id, **params)#

Search a merchant's product catalog.

result = await sc.catalog.search(
    "mrc_660e8400...",
    q="organic bananas",
    category="groceries",
    limit=10,
)

for item in result["items"]:
    print(item["name"], item["price"])

CatalogSearchParams:

| Field | Type | Required | Description | |---|---|---|---| | q | str | No | Search query | | category | str | No | Filter by category | | limit | int | No | Number of results | | cursor | str | No | Pagination cursor |

LangChain Integration#

The Python SDK includes two LangChain tools for AI agents. Install the extra:

pip install subcent[langchain]

SubcentPaymentTool#

A LangChain BaseTool that allows an agent to make payments.

from subcent import Subcent
from subcent.integrations.langchain import SubcentPaymentTool

sc = Subcent(api_key="sc_agent_your_agent_key", base_url="https://testnet-api.subcent.io")

payment_tool = SubcentPaymentTool(client=sc)

# Use in a LangChain agent
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")
tools = [payment_tool]

# The tool accepts these parameters:
# - agent_id (str): The agent ID making the payment
# - merchant_id (str): The merchant ID to pay
# - amount (int): Amount in minor units
# - currency (str): Payment currency (default: "USDC")
# - category (str): Payment category
# - memo (str, optional): Payment memo

Tool schema:

| Field | Type | Required | Description | |---|---|---|---| | agent_id | str | Yes | The agent ID making the payment | | merchant_id | str | Yes | The merchant ID to pay | | amount | int | Yes | Amount in minor units (e.g., cents) | | currency | str | No | Payment currency (default: "USDC") | | category | str | Yes | Payment category | | memo | str | No | Optional payment memo |

The tool returns a string like: "Payment pay_abc123 created with status: completed"

SubcentCatalogTool#

A LangChain BaseTool that allows an agent to search merchant catalogs.

from subcent.integrations.langchain import SubcentCatalogTool

catalog_tool = SubcentCatalogTool(client=sc)

# The tool accepts these parameters:
# - merchant_id (str): Merchant ID to search
# - query (str, optional): Search query

Tool schema:

| Field | Type | Required | Description | |---|---|---|---| | merchant_id | str | Yes | Merchant ID to search | | query | str | No | Search query |

The tool returns a string like: "Found 3 items: Organic Bananas, Fresh Milk, Whole Wheat Bread"

Full LangChain Example#

import asyncio
from subcent import Subcent
from subcent.integrations.langchain import SubcentPaymentTool, SubcentCatalogTool
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

async def main():
    sc = Subcent(
        api_key="sc_agent_your_agent_key",
        base_url="https://testnet-api.subcent.io",
    )

    tools = [
        SubcentPaymentTool(client=sc),
        SubcentCatalogTool(client=sc),
    ]

    llm = ChatOpenAI(model="gpt-4")
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a shopping assistant. Use the tools to search catalogs and make payments."),
        MessagesPlaceholder(variable_name="chat_history", optional=True),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ])

    agent = create_openai_functions_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    result = await executor.ainvoke({
        "input": "Find organic bananas from merchant mrc_freshmart and buy 2 bunches"
    })
    print(result["output"])

    await sc.close()

asyncio.run(main())

Error Handling#

The SDK raises ApiError for all HTTP error responses.

from subcent.errors import ApiError, SubcentError

try:
    payment = await sc.payments.request({
        "agent_id": "agt_invalid",
        "merchant_id": "mrc_660e8400...",
        "amount": 2500,
        "currency": "USDC",
        "category": "groceries",
    })
except ApiError as e:
    print(e.status_code)   # 403
    print(e.error_code)    # "agent_not_authorized"
    print(e.message)       # "Agent not found"
    print(e.details)       # additional error details if available
except SubcentError as e:
    print(e.message)       # base error class

Error Classes#

| Class | Description | |---|---| | SubcentError | Base exception class with a message attribute | | ApiError | Raised on HTTP error responses. Includes status_code, error_code, message, and details |