Commerce infrastructure for AI agents in Latin America. Pix, NF-e, WhatsApp, shipping, banking — one API, no provider-key boilerplate.
pip install codesparPython 3.10+ required.
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 +5511999887766 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)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(...)).
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": "alice@example.com"}, "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.
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")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())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.
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.
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.
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.
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.
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:8000Then your code stays unchanged:
cs = CodeSpar(api_key="local") # OSS runtimes accept any non-empty bearerPass base_url= explicitly when you need to override the env var.
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:
raiseThe 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.
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 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.