Get a CodeSpar commerce agent running in Python — sync for scripts and Django, async for FastAPI and LangChain. Under 5 minutes.

Quickstart (Python)

codesparv0.9.0

The codespar PyPI package exposes the same session model as the TypeScript SDK, in both sync and async flavours. Use CodeSpar from scripts, Jupyter notebooks, Flask, or Django views. Use AsyncCodeSpar from FastAPI, LangChain, or anything already on asyncio.

Prerequisites

Python 3.10+
A CodeSpar API key — mint one at Dashboard → API Keys. New accounts get a test-environment project plus a csk_test_* key auto-created at signup.
Optional: an LLM SDK (anthropic, openai) if you want to build a full agent loop

Install

pip install codespar

Or add to pyproject.toml:

[project]
dependencies = ["codespar>=0.9.0"]

The package has a single runtime dependency (httpx). No async framework is pulled in — both flavours ship in the same wheel.

Set your API key

export CODESPAR_API_KEY="csk_test_your_key_here"

A csk_test_* key authorizes against a test-environment project. Tool calls route to the catalog's base_url_test provider sandbox where one is defined, and to the real production endpoint otherwise. For deterministic responses without any upstream call, declare per-session fixtures via cs.create({ mocks: {...} }) — see Test Mode for the full reference.

Run a session against an inline mock

The fastest way to confirm the SDK is wired correctly — and the pattern most tests use — is to declare a mock on session create. The runtime substitutes the fixture for the upstream call; policy, audit, and commerce-memory still fire on every invocation.

test_mode.py

# Requires codespar==0.10.0+
from codespar import CodeSpar
import os

cs = CodeSpar(api_key=os.environ["CODESPAR_API_KEY"])

with cs.create("user_test", {
    "servers": ["asaas"],
    "mocks": {
        "asaas/create_payment": {"id": "pay_test_42", "status": "PENDING"},
    },
}) as session:
    result = session.execute("asaas/create_payment", {"value": 100})
    print(result["id"])  # "pay_test_42"

cs.close()

test_mode.py

# Requires codespar==0.10.0+
import asyncio
import os
from codespar import AsyncCodeSpar

async def main():
    async with AsyncCodeSpar(api_key=os.environ["CODESPAR_API_KEY"]) as cs:
        async with await cs.create("user_test", {
            "servers": ["asaas"],
            "mocks": {
                "asaas/create_payment": {"id": "pay_test_42", "status": "PENDING"},
            },
        }) as session:
            result = await session.execute("asaas/create_payment", {"value": 100})
            print(result["id"])  # "pay_test_42"

asyncio.run(main())

A non-empty mocks field puts the session into strict mode for its lifetime: any tool call whose canonical name is not declared returns a tool_not_mocked envelope rather than falling through to the real upstream. See Test Mode for the strict-mode contract, the five tool_result envelopes, and how to assert on them in Python (is_policy_denied, is_approval_required, is_mocks_exhausted, is_tool_not_mocked).

Create a session + call a tool

charge.py

from codespar import CodeSpar

cs = CodeSpar()  # reads CODESPAR_API_KEY from env

with cs.create("user_123", preset="brazilian") as session:
    result = session.execute(
        "codespar_pay",
        {"method": "pix", "amount": 15000, "currency": "BRL"},
    )
    print(result.data["pix_code"])

cs.close()

Context-manager style closes the session automatically. Or call session.close() and cs.close() manually.

charge.py

import asyncio
from codespar import AsyncCodeSpar

async def main():
    async with AsyncCodeSpar() as cs:
        async with await cs.create("user_123", preset="brazilian") as session:
            result = await session.execute(
                "codespar_pay",
                {"method": "pix", "amount": 15000, "currency": "BRL"},
            )
            print(result.data["pix_code"])

asyncio.run(main())

Natural-language agent loop

session.send(message) runs a Claude tool-use loop on the backend. No local agent framework needed.

agent.py

from codespar import CodeSpar

cs = CodeSpar()
with cs.create("user_123", preset="brazilian") as session:
    result = session.send(
        "Charge R$150 via Pix for order #5678 and send a WhatsApp confirmation"
    )
    print(result.message)
    for call in result.tool_calls:
        print(f"  → {call.tool_name}: {call.status} in {call.duration_ms}ms")
cs.close()

agent.py

import asyncio
from codespar import AsyncCodeSpar

async def main():
    async with AsyncCodeSpar() as cs:
        async with await cs.create("user_123", preset="brazilian") as session:
            result = await session.send(
                "Charge R$150 via Pix for order #5678 and send a WhatsApp confirmation"
            )
            print(result.message)

asyncio.run(main())

Streaming

send_stream yields events as they happen — useful when you want to stream partial output to the user or to a log.

for event in session.send_stream("Charge R$150 via Pix"):
    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 == "done":
        print(f"\nFinal: {event.result.message}")

async for event in session.send_stream("Charge R$150 via Pix"):
    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 == "done":
        print(f"\nFinal: {event.result.message}")

Import surfaces

Class	Use from
`CodeSpar`	Sync scripts, Jupyter, Flask / Django views
`AsyncCodeSpar`	FastAPI, async LangChain, anything on asyncio
`Session` / `AsyncSession`	Returned by `create()` — same method names, sync vs async dispatch

Both clients wrap the same REST API on api.codespar.dev. Start with sync, upgrade to async without changing the surrounding code — method names and return types match.

Common errors

Error class	Cause	Fix
`ConfigError`	Missing or malformed `api_key`	Set `CODESPAR_API_KEY` or pass `CodeSpar(api_key="csk_...")`
`NotConnectedError`	Calling a tool before a server finishes connecting	Use `manage_connections={"wait_for_connections": True}` in `create()`
`ApiError`	Backend returned non-2xx	Inspect `error.code` and `error.message`; retry or surface to user
`StreamError`	SSE stream dropped mid-flight	Reconnect; partial results are in the final event if already emitted