Quickstart
This page takes you from pip install to a small working app: an agent,
a tool it can call, a typed result, live streaming, and a workflow that
routes between two agents. Every block is a complete script. Save it as
a .py file and run it with python.
Step 1 runs without an API key. The others need an Anthropic or OpenAI key.
Setup
Bare install
pip install loomflowEnough for step 1. The built-in echo model runs the whole agent
loop without calling any provider.
1. Hello, agent
import asyncio
from loomflow import Agent
async def main():
agent = Agent("You are a helpful assistant.", model="echo")
result = await agent.run("Tell me a joke.")
print(result.output)
asyncio.run(main())An Agent takes instructions and a model. model="echo" repeats the
prompt back, so you can see the loop work before spending tokens.
model= is required. There is no default model; leaving it out
raises a ConfigError that lists the values you can pass.
run() returns a RunResult. Besides output it carries turns,
tokens_in, tokens_out, cost_usd and session_id.
2. Use a real model
Change the model string. The prefix picks the provider:
import asyncio
from loomflow import Agent
async def main():
agent = Agent("You are a helpful assistant.", model="claude-opus-4-7")
# model="gpt-4o" would use OpenAI instead
result = await agent.run("In one sentence, what is a context window?")
print(result.output)
print(f"{result.tokens_in} tokens in, {result.tokens_out} out, ${result.cost_usd:.4f}")
asyncio.run(main())claude-* goes to Anthropic, gpt-*, o1, o3 and
o1-*/o3-*/o4-* to OpenAI, deepseek-* to DeepSeek, and prefixes such as gemini/,
ollama/ or bedrock/ go through LiteLLM. To set options like
max_tokens, pass a model object instead of a string:
from loomflow import Agent
from loomflow.model.anthropic import AnthropicModel
agent = Agent(
"You are a helpful assistant.",
model=AnthropicModel("claude-opus-4-7", max_tokens=8192),
)Models covers every provider, fallbacks and timeouts.
3. Give it a tool
Decorate a function with @tool and pass it to the agent. The
docstring and type hints become the tool definition the model sees:
the docstring’s text before its first section (Args:, Returns:, …) is the
description, and the Args: entries describe the parameters.
import asyncio
from loomflow import Agent, tool
@tool
async def get_weather(city: str) -> str:
"""Look up the current weather for a city."""
return f"Sunny, 22°C in {city}." # call a real weather API here
async def main():
agent = Agent(
"You answer questions about the weather.",
model="claude-opus-4-7",
tools=[get_weather],
)
result = await agent.run("Should I bring a jacket in Tokyo today?")
print(result.output)
asyncio.run(main())The agent decides when to call the tool, reads the result and keeps
going until it can answer. Plain (non-async) functions work too; they
run on a worker thread so they don’t block. Mark tools that change
things with @tool(destructive=True) so the default permissions ask
before running them (Permissions).
4. Get a typed result
Pass a Pydantic model as output_schema= and get a validated object
back instead of text:
import asyncio
from pydantic import BaseModel
from loomflow import Agent
class WeatherReport(BaseModel):
city: str
temp_c: float
conditions: str
async def main():
agent = Agent("Extract the facts exactly.", model="gpt-4o")
result = await agent.run(
"Tokyo today: sunny, 22 degrees Celsius.",
output_schema=WeatherReport,
)
report = result.parsed # a WeatherReport
print(report.city, report.temp_c, report.conditions)
asyncio.run(main())If the model’s answer doesn’t match the schema, loomflow tells it what
was wrong and asks again. You can also set output_schema= once on
the Agent so every run uses it, or pass a union such as
Found | NotFound to allow more than one shape. Details are in the
Agent reference.
5. Stream the answer
agent.stream() yields events while the run is happening, so you can
show text as it arrives:
import asyncio
from loomflow import Agent
async def main():
agent = Agent("You are a travel planner.", model="claude-opus-4-7")
async for event in agent.stream("Plan one day in Tokyo."):
if event.kind == "model_chunk":
chunk = event.payload["chunk"]
if chunk["kind"] == "text":
print(chunk["text"], end="", flush=True)
elif event.kind == "tool_call":
print(f"\n[calling {event.payload['call']['tool']}]")
print()
asyncio.run(main())Other event kinds include started, tool_result, budget_warning,
error and completed. Streaming UI
shows how to send them to a browser.
6. Route between agents with a workflow
An agent lets the model decide the steps. A Workflow is the
opposite: you decide the steps, and agents can be steps. Here a small
model sorts each message, then the right specialist answers it:
import asyncio
from loomflow import Agent, Workflow
billing = Agent("You handle billing questions.", model="claude-opus-4-7")
tech = Agent("You handle technical questions.", model="claude-opus-4-7")
async def classify(text: str) -> str:
result = await Agent(
"Reply with one word: billing or tech.",
model="claude-haiku-4-5",
).run(text)
return "billing" if "billing" in result.output.lower() else "tech"
support = Workflow.route(classify, {"billing": billing, "tech": tech})
async def main():
result = await support.run("My card was charged twice.", user_id="alice")
print(result.visited) # ['classify', 'route_billing']
print(result.output)
asyncio.run(main())classify returns a key, and Workflow.route sends the message to
the agent under that key. (It checks the reply rather than trusting it
verbatim, because a model may answer “Billing.” with a capital and a
full stop.) result.visited lists the steps that ran.
user_id keeps each user’s memory, budget and audit records separate.
Workflow covers chain, parallel, loops and
building graphs by hand; Agent vs Workflow
helps you choose.
Next steps
You now have the core of loomflow. The Production tour adds the pieces a real deployment needs, one short example each: memory, crash recovery, telemetry, the audit log, permissions, sandboxing and budgets.