codespar 0.10.0

Python SDK for CodeSpar — commerce infrastructure for AI agents in Latin America.

codespar — Python SDK

Commerce infrastructure for AI agents in Latin America. Pix, NF-e, WhatsApp, shipping, banking — one API, no provider-key boilerplate.

Install

pip install codespar

Python 3.10+ required.

Quick start

from codespar import CodeSpar

cs = CodeSpar(api_key="csk_live_...")

session = cs.create(
    "user_123",
    preset="brazilian",   # zoop, nuvem-fiscal, melhor-envio, z-api, omie
    # project_id="prj_...", # optional — defaults to the org's default project
)

result = session.send(
    "Charge R$500 via Pix to [phone] and send the QR code by WhatsApp."
)
print(result.message)
for call in result.tool_calls:
    print(f"  → {call.tool_name} ({call.duration_ms}ms)")

session.close()
cs.close()

Or as a context manager:

with CodeSpar(api_key="csk_live_...") as cs:
    session = cs.create("user_123", preset="brazilian")
    print(session.send("Quero pagar R$125 via Pix").message)

Tool discovery + connection wizard

Beyond session.execute(tool, params), the SDK exposes typed wrappers for the codespar_discover and codespar_manage_connections meta-tools:

from codespar import CodeSpar, ConnectionWizardOptions, DiscoverOptions

with CodeSpar(api_key="csk_live_...") as cs:
    session = cs.create("user_123", preset="brazilian")

    # Find the right tool for a free-form use case.
    found = session.discover(
        "send a pix payment",
        DiscoverOptions(country="BR", limit=3),
    )
    if found.recommended:
        print(found.recommended.server_id, found.recommended.tool_name)
        print(f"  status: {found.recommended.connection_status}")

    # Surface the connection wizard if the recommended server isn't
    # connected. NEVER pass credentials through this method —
    # credentials only travel via the dashboard's connect modal or
    # the OAuth callback. The wizard returns a deep-link the agent
    # surfaces so the user finishes setup in their browser.
    if found.recommended and found.recommended.connection_status == "disconnected":
        wiz = session.connection_wizard(
            ConnectionWizardOptions(
                action="initiate",
                server_id=found.recommended.server_id,
            ),
        )
        if wiz.initiate:
            print("Connect:", wiz.initiate.connect_url)
            for line in wiz.initiate.instructions:
                print(" ·", line)

Async users have the same surface on AsyncSession (await session.discover(...), await session.connection_wizard(...)).

Meta-tool wrappers + async settlement

In addition to discover / connection_wizard, Session (sync) and AsyncSession exposes typed wrappers for charges, shipping, and async settlement / verification polling:

from codespar import CodeSpar

with CodeSpar(api_key="csk_live_...") as cs:
    session = cs.create("user_123", preset="brazilian")

    # Inbound charge — buyer pays merchant. Pix BRL routes via
    # Asaas / MP / iugu / Stone with failover.
    charge = session.charge(
        amount=150,
        currency="BRL",
        method="pix",
        buyer={"name": "Cliente Demo", "document": "11144477735"},
    )

    # Shipping label via Melhor Envio (action="label"|"quote"|"track")
    label = session.ship(
        action="label",
        origin={...},
        destination={...},
        items=[...],
    )

    # Async settlement — codespar_charge returns synchronously, but real
    # settlement lands via webhook. Poll, or stream over SSE.
    settled = session.payment_status(charge.tool_call_id)

    def on_payment(env):
        print("payment status →", env.status)

    session.payment_status_stream(
        charge.tool_call_id,
        on_update=on_payment,  # sync or async callable
    )

    # Async KYC — codespar_kyc returns the inquiry id; the buyer
    # finishes the hosted flow off-platform.
    inquiry = session.execute(
        "codespar_kyc",
        {"buyer": {"email": "[e-mail]"}, "check_type": "identity"},
    )
    v = session.verification_status(inquiry.tool_call_id)
    #   approved | rejected | review | expired | pending

    session.verification_status_stream(
        inquiry.tool_call_id,
        on_update=lambda env: print("kyc status →", env.status),
    )

Both streaming methods return the last envelope when the backend closes (typically 5s after a terminal state). Cancel from the caller side by wrapping in an asyncio.Task and calling .cancel() on AsyncSession; for the sync Session the stream returns when the backend tears down (terminal + 5s) or you raise from on_update.

Every method listed above exists on both Session (sync) and AsyncSession (await the async variants). Snake-case throughout — TS paymentStatusStream ⇆ Python payment_status_stream.

Streaming

for event in session.send_stream("Process order #BR-7721"):
    if event.type == "assistant_text":
        print(event.content, end="", flush=True)
    elif event.type == "tool_use":
        print(f"\n→ calling {event.name}...")
    elif event.type == "tool_result":
        print(f"  {event.tool_call.status} in {event.tool_call.duration_ms}ms")

Async

import asyncio
from codespar import AsyncCodeSpar

async def main():
    async with AsyncCodeSpar(api_key="csk_live_...") as cs:
        session = await cs.create("user_123", preset="brazilian")
        result = await session.send("charge R$500 via Pix")
        print(result.message)
        await session.close()

asyncio.run(main())

Multi-environment (projects)

CodeSpar scopes every session to an environment (prj_<id>). Pass a project id on the client for the whole lifetime, or per-session when you want to target a different environment:

# Pin every session this client spawns to the staging project
cs = CodeSpar(api_key="csk_live_...", project_id="prj_staging0123abcd")

# Override per session
session = cs.create("user_123", preset="brazilian", project_id="prj_prod0123abcd")

When you omit project_id, CodeSpar routes to the org's default project — always defined, self-healed on first read.

Raw HTTP proxy

Skip the agent loop and hit a provider API directly through CodeSpar's credential vault:

from codespar import ProxyRequest

response = session.proxy_execute(ProxyRequest(
    server="stripe-acp",
    endpoint="/v1/charges",
    method="POST",
    body={"amount": 2000, "currency": "brl"},
))
print(response.status, response.data)

No API key leaves your machine — CodeSpar injects it server-side.

Connect Links (OAuth)

from codespar import AuthConfig

link = session.authorize(
    "stripe-acp",
    AuthConfig(redirect_uri="https://your.app/connected"),
)
print(f"Open this URL to connect Stripe: {link.authorize_url}")

After the user completes the OAuth flow, CodeSpar stores the tokens in the per-project vault and forwards them back to redirect_uri with ?status=connected&connection_id=<id> appended.

Test-mode mocks

Skip live providers in tests by passing a mocks dict to cs.create. Keys are canonical tool names in slash form (asaas/create_payment, melhor-envio/calculate_shipping, ...). Values are either a single dict — used as the response on every matching call — or a list of dicts consumed in order, returning mocks_exhausted once the list drains.

import os
from codespar import CodeSpar

with CodeSpar(api_key=os.environ["CODESPAR_API_KEY"]) as cs:
    session = cs.create(
        "user_test",
        servers=["asaas"],
        mocks={
            "asaas/create_payment": {"id": "pay_test", "status": "PENDING"},
        },
    )

    result = session.execute("asaas/create_payment", {"value": 100})
    # result.data == {"id": "pay_test", "status": "PENDING"}

Pass a list for stateful mocks:

mocks={
    "asaas/create_payment": [
        {"id": "pay_1", "status": "PENDING"},
        {"id": "pay_1", "status": "RECEIVED"},
    ],
}

Mocks live behind the managed backend's test-mode gate — a csk_test_* API key against a test-environment project. Live keys against the same map return mocks_not_permitted. The SDK forwards keys verbatim; the OSS double-underscore form (asaas__create_payment) reaches the backend unrewritten and surfaces as mocks_invalid rather than the SDK silently rewriting.

The OSS runtime accepts the same mocks shape on its session API (see codespar/codespar#113), so the same fixtures work whether you point at api.codespar.dev or a self-hosted instance via CODESPAR_BASE_URL. Self-hosted runtimes must additionally set CODESPAR_TEST_MODE_ENABLED=true on the server process; without it, the SDK receives mocks_not_permitted / HTTP 501 instead of fixture responses.

Storage shape differs by runtime — the wire contract does not. The managed backend persists mocks and per-tool consume counters; sessions and their fixtures survive restarts and multi-replica deployments. The OSS runtime holds both in process memory; they are scoped to the HTTP-session process and are lost on restart, and channel-bridge sessions (WhatsApp, Slack, Telegram, Discord) cannot carry mocks under the OSS shape. Response envelopes, status codes, sibling fields, and gate ordering are byte-identical between runtimes regardless. See the test-mode concept doc for the full per-runtime split.

Test mode is a property of the runtime, not the session. On the managed backend it's project.environment == 'test'; on a self-hosted OSS runtime it's CODESPAR_TEST_MODE_ENABLED=true on the server process. When the runtime is in test mode, every external tool call your code or LLM dispatches must match a declared mock — unmatched calls return tool_not_mocked (HTTP 422 on the catalog-routed /execute path; a tool_result block on the chat-loop) and no upstream provider runs. The envelope covers three failure modes: the mocks map has no entry for the canonical name, the session was created with no mocks field, or the canonical name has an unknown server prefix. A session that doesn't declare mocks can't dispatch any tools in test mode; declare the mocks the test will exercise, or run the same code against a live-mode runtime where the real providers handle dispatch. Built-in metadata tools — codespar_list_tools on OSS, codespar_discover and codespar_manage_connections on the managed backend — bypass this gate.

Type aliases

MockObject (dict[str, Any]) and MockValue (MockObject | list[MockObject]) export from codespar. Use them when you want to define fixtures separately from the create call site.

from codespar import MockValue

fixtures: dict[str, MockValue] = {
    "asaas/create_payment": {"id": "pay_test", "status": "PENDING"},
}

AsyncCodeSpar accepts the same mocks=... kwarg.

Base URL — managed or self-hosted OSS

CodeSpar and AsyncCodeSpar honor the CODESPAR_BASE_URL environment variable. The constructor cascade is explicit base_url arg, then the env var, then the managed default at https://api.codespar.dev.

# Point the same client wiring at a local OSS runtime
export CODESPAR_BASE_URL=http://localhost:8000

Then your code stays unchanged:

cs = CodeSpar(api_key="local")  # OSS runtimes accept any non-empty bearer

Pass base_url= explicitly when you need to override the env var.

Errors

Every failure wraps in the CodeSparError hierarchy:

from codespar import ApiError, ConfigError, StreamError

try:
    session = cs.create("user_123", preset="brazilian")
except ConfigError as exc:
    print(f"Bad config: {exc}")
except ApiError as exc:
    print(f"Backend said {exc.status}: {exc.code}")

ApiError.code is the structured discriminant on every non-success response — branch on it instead of parsing str(exc) or exc.body. The legacy error field on pre-test-mode envelopes is honored as a fallback when code is missing, so older responses still surface a code value.

from codespar import ApiError, CodeSpar

try:
    cs.create("user_test", mocks={"asaas/create_payment": {}})
except ApiError as exc:
    if exc.code == "mocks_not_permitted":
        # Live key against a mocks map. Swap to csk_test_*.
        ...
    elif exc.code == "mocks_invalid":
        # Backend rejected a tool-name key. Check the slash form.
        ...
    elif exc.status == 0:
        # Network never reached the backend; exc.__cause__ has the httpx error.
        ...
    else:
        raise

Tool-result guards

The five reserved tool-result codes ship typed guards plus an exhaustive-match helper. Each guard checks the code discriminant AND the variant's required sibling fields — a payload with a well-formed code but a missing rule_id / approval_id / tool_name returns False rather than narrowing positive.

from codespar import (
    CodeSpar,
    assert_exhaustive_tool_result,
    is_approval_required,
    is_mocks_engine_error,
    is_mocks_exhausted,
    is_policy_denied,
    is_tool_not_mocked,
)

with CodeSpar(api_key="csk_test_...") as cs:
    session = cs.create("user_test", servers=["asaas"])
    result = session.execute("asaas/create_payment", {"value": 100})

    if is_policy_denied(result.data):
        print(f"blocked by {result.data['rule_id']}: {result.data['message']}")
    elif is_approval_required(result.data):
        print(f"approval {result.data['approval_id']} expires {result.data['expires_at']}")
    elif is_mocks_exhausted(result.data):
        # Stateful mock list drained — pad it or extend the test.
        ...
    elif is_mocks_engine_error(result.data):
        # Backend-side mocks engine failure; usually a malformed fixture.
        ...
    elif is_tool_not_mocked(result.data):
        print(f"no mock for {result.data['tool_name']}")

assert_exhaustive_tool_result(value) raises AssertionError from a default branch — call it after handling each variant so a sixth code landing without a handler trips at the boundary instead of being silently swallowed.

Design parity with the JS SDK

This package mirrors @codespar/sdk method-for-method. Same backend, same payloads, same preset names — pick the language that fits your stack without giving anything up. Every 0.9.0 / 0.7.0 method on the JS Session (charge, ship, payment_status, payment_status_stream, verification_status, verification_status_stream, discover, connection_wizard) exists on the Python Session and AsyncSession with snake_case naming.

Need more?

Need governance, budget limits, and audit trails for agent payments? CodeSpar Enterprise adds policy engine, payment routing, and compliance templates on top of these MCP servers.

Links

• Documentation
• Dashboard
• JS SDK (npm)
• Report a bug