composeai — 𝑓(𝑔(𝑥))

A radically simple, functional framework for multi-agent AI workflows.

• Agents are typed functions. The docstring is the system prompt, the body returns the user prompt, the return annotation is the structured output type. Calling one returns that type — or raises.
• Composition is checked before it runs. pipe(researcher, copywriter) verifies every stage boundary at build time; a wiring bug raises CompositionTypeError before a single API call is made.
• Tracing is always on, and local. Every run persists spans, token usage, and cost to a SQLite store on your disk (./.compose). No SaaS, no instrumentation, no opt-in — the trace is just there.
• Flows are durable. A @flow journals every step; if it crashes — or pauses on a named interrupt (approve("publish")) waiting for a human — resume(run_id) continues it in the same process or a brand-new one, days later, replaying finished steps without re-paying for them.
• Every run can carry a spend cap. Budget(usd=..., tokens=...) is enforced after every LLM call in a run's subtree, and stays cumulative across resume() — a run can't dodge its budget by crashing and getting resumed.
• Streaming and tracing are the same event bus. .stream() yields text_delta/thinking_delta/tool_call_started/tool_args_delta interleaved with the very span_started/span_finished/run_finished events the trace is built from — on agents, pipelines, and flows alike, so a live UI and the trace can never disagree.

Runtime dependencies: pydantic + the standard library. Provider SDKs are optional extras. Python ≥ 3.10.

Install

pip install composeai                  # core: pydantic + stdlib only
pip install "composeai[anthropic]"     # + the Anthropic SDK
pip install "composeai[openai]"        # + the OpenAI SDK
pip install "composeai[all]"           # both

The 90-second tour

Define an agent as a typed function. The docstring is its system prompt, the return annotation is its structured output type:

import composeai as compose
from pydantic import BaseModel, Field


class FactSheet(BaseModel):
    topic: str
    key_facts: list[str] = Field(description="Three crisp, verifiable facts")


@compose.tool
def count_words(text: str) -> int:
    """Count the words in a piece of text.

    Args:
        text: The text whose words should be counted.
    """
    return len(text.split())


@compose.agent(model="anthropic/claude-sonnet-5", tools=[count_words])
def researcher(topic: str) -> FactSheet:
    """You are a meticulous researcher. Extract crisp, verifiable facts."""
    return compose.prompt(f"Build a fact sheet about: {topic}")


facts = researcher("quantum computing")   # -> FactSheet (or raises)

Calling researcher(...) runs the whole loop and returns the validated FactSheet. Every call like this is automatically persisted, so you can inspect it from the command line right afterward:

$ compose trace --last
trace 01KXC6RDB8E29NZHEAC2F54M11 — ok — [$0.0150 · 2.9k tok · 3ms]
└─ ◆ researcher [$0.0150 · 2.9k tok · 3ms]
   ├─ ▸ anthropic/claude-sonnet-5 [$0.0075 · 1.5k tok · 0ms]
   ├─ ⚙ count_words [0ms]
   └─ ▸ anthropic/claude-sonnet-5 [$0.0075 · 1.5k tok · 0ms]

No accounts, no exporters, no instrumentation to wire up — the trace (and its cost) is just there, in ./.compose, the moment the agent finishes. See agents for .run()/.stream(), repairs, and resilience knobs; composition and flows for wiring agents into pipelines and durable, resumable workflows; budgets for the Budget(usd=..., tokens=...) cap shown above.

Documentation

Rules of the road

The contracts composeai holds you to — and the ones it holds itself to:

• Flow bodies must be deterministic. Replay works by re-running the body and substituting journaled step results in call order. Side effects, randomness, and clock reads belong inside @task/@agent steps, never in the flow body itself. Nothing detects a violation; it just won't replay correctly.
• A step journals only after it returns. If the process dies between a task's external side effect and its journal write, resume re-runs that task — make external side effects idempotent.
• Journals are for pause/approval and crash recovery, not cross-release storage. If a @flow's source changes between pause and resume, resume() fails loud with ResumeMismatchError (the journal may no longer match the new call sequence); allow_code_change=True overrides it deliberately.
• State lives at COMPOSE_DIR (default ./.compose). COMPOSE_TRACE_CONTENT=0 stops spans from capturing input/output payloads — usage, status, and timing are always recorded regardless. This does not extend to anything composeai needs as functional state to actually work, all of which are written in full unconditionally, regardless of COMPOSE_TRACE_CONTENT: a paused agent's agent_state snapshot (durable pause/resume — approve()/ask_human()/@tool(requires_approval=True) — requires the full in-progress conversation, including tool call arguments and results, so resume() can continue exactly where it left off); the @flow journal (step results must be real to replay correctly); and the test kit's own on-disk artifacts — record_cassette/the cassette fixture and @compose.agent(cache=True)'s filesystem response cache both need a call's real request/response to be replayable or servable as a cache hit later. If your tools handle secrets/PII in their arguments or results, treat {COMPOSE_DIR} (runs.db, cache/, and any cassette files you commit) as sensitive (filesystem permissions, encryption at rest, etc.) rather than relying on COMPOSE_TRACE_CONTENT to keep it out of the store.
• temperature is passthrough-only. composeai never sets one for you, and modern Claude models reject sampling parameters with a 400 — leave it unset for Claude.
• Cost is never fabricated. Priced models get exact USD from a dated, in-repo price table; calls with no known price report cost_usd=None, and any total that includes them renders as a ≥$X partial (or - when nothing was priceable) instead of a made-up number. USD budgets consequently can't see unpriced spend — set a token budget too if you need a hard cap.
• Retries can re-stream. With retries > 0, a provider error striking mid-stream re-runs the call from the start — consumers of .stream() may see the same deltas twice on one llm span. Render final outputs (or treat a fresh delta burst as a reset) if double-rendering matters.

vs. the alternatives

Releasing (maintainers)

Two equivalent paths:

• Local: put a project-scoped PyPI token in .env (git-ignored; see the placeholder), commit your work, then scripts/release.sh X.Y.Z — it bumps the version, runs the full gate, builds, and uploads. Commit the bump and tag vX.Y.Z afterwards.
• GitHub: publish a GitHub Release and .github/workflows/release.yml tests, builds, and publishes via PyPI trusted publishing (no stored token). One-time setup: add this repo as a trusted publisher on pypi.org (workflow release.yml, environment pypi).

Roadmap

• OpenTelemetry exporter (the span model already tracks gen_ai.* attribute conventions)
• Async API (await agent(...), async tools)
• TypeScript sibling package
• Extended-thinking / reasoning request configuration (Anthropic thinking budget, OpenAI reasoning.summary/encrypted_content) -- today ThinkingPart only round-trips whatever a provider returns unprompted by default; there's no ModelRequest field to actually ask for it

License

MIT