Loomflow

A Python framework for building agents and workflows that ship — two primitives held in tension, like the warp and weft of a loom. Workflow is the warp: the graph you hold taut. Agent is the weft: the loop you hand to the model. They share one spine for telemetry, audit, and user_id partitioning. Typed outputs, MCP support, swap models with one argument.

import asyncio
from loomflow import Agent, Workflow
 
# 1) Agent — LLM-controlled loop
billing = Agent("Handle billing.", model="gpt-4.1-mini", tools=[...])
tech    = Agent("Handle tech.",    model="gpt-4.1-mini", tools=[...])
 
# 2) Workflow — developer-controlled DAG
async def classify(text: str) -> str:
    return (await Agent(
        "Reply 'billing' or 'tech'.", model="gpt-4.1-mini",
    ).run(text)).output
 
support = Workflow.route(
    classify,
    {"billing": billing, "tech": tech},
)
 
async def main():
    r = await support.run("My card was charged twice.", user_id="alice")
    print(r.output)
    print(r.visited)        # ['classify', 'route_billing']
 
asyncio.run(main())

What’s new in this release

The current release lands durability, security, and serving in one sweep. Every item below has a full page:

• Durable checkpoints — Tuning(checkpoint=True) snapshots the transcript every pass; agent.resume(..., from_checkpoint=...) restores or forks a session without re-billing completed turns.
• Signals — park a run on an external event, resume it later with agent.signal(...). The backbone of long-lived human-in-the-loop flows.
• Guardrails — injection, PII, moderation, and regex guards at the input / output / tool-result stages. Zero cost when unset.
• Richer approvals — ApprovalDecision adds edit-and-approve and per-run approval memory on top of the bool handlers.
• OTel gen_ai.* conventions — spans now carry the GenAI semantic-convention attributes, so Langfuse / Datadog / Phoenix-class backends work out of the box.
• Rate limiting — client-side TokenBucketRateLimiter, per-user or global, throttle or raise.
• Serving — loomflow.serve.create_app exposes any agent over HTTP (/run, /stream SSE, /resume), plus the loom CLI.
• A2A — publish an agent card, consume a remote agent as a tool with A2AClient(...).as_tool().
• Evals — EvalHarness with built-in metrics, LLM-judge scoring, CI threshold gating, online sampling.
• Fact graph — multi-hop recall_graph over bi-temporal facts, point-in-time traversal, entity merging. Plus a token budget for what recall injects into the prompt.
• Tool search and code mode — scale past large tool inventories: deferred tool loading, and tools-as-Python-API.
• Model resilience — fallback chains across providers and per-request timeouts.
• Per-role model routing — plan with a frontier model, execute with a cheap one, inside one agent: PlanAndExecute(planner_model=..., executor_model=...), plus <role>_model= on ToT / ReWOO / Reflexion / SelfRefine.
• Parallel workflows — richer fan-out / fan-in, and supervisor spawning for dynamic subagents. MCP support also grew.

Examples 24–31 in the repo walk the new surface end-to-end — see Examples.

Why pick this

• Two primitives, one spine. Agent and Workflow are siblings. Same telemetry, same audit log, same user_id partition no matter which one you pick. Pass an Agent as a Workflow node, or call wf.as_tool() and the workflow becomes a tool the agent can call.
• user_id is a typed primitive. One Agent plus one Memory partitions cleanly across N tenants. Namespacing is automatic.
• output_schema= accepts any Pydantic model. The framework appends the schema directive to the prompt, parses the response, validates it, and retries with feedback if the model misses. You get a typed instance back.
• Network model adapters get wrapped with a typed error taxonomy and a retry policy. Rate limits, 5xx, network blips don’t blow up your run.
• session_id rehydrates prior turns as real chat history. Reuse the same id across processes and the journal replays.
• Twelve agent-loop architectures ship behind one Agent constructor. One kwarg flips the iteration pattern.

Where to start

Install

pip install loomflow
 
# With provider SDKs
pip install 'loomflow[anthropic,openai]'

For zero-key local development, the bare install ships an EchoModel that lets you exercise the full loop without burning tokens.