weaveflow 2.0.0

Weave: composable AI agents with a universal interface. Connect any framework (LangChain, CrewAI, or any Python callable) with no rewrite.

Weave: composable AI agent framework

USB for AI agents. Build an agent once against an open contract, and connect it to any other compliant agent, regardless of LLM, language, or host.

Weave is a Python framework for building composable AI agents. Every agent exposes typed input/output ports and capability tags as its public interface; its brain (LLM), memory, and tools stay private. Any agent's output can plug into any compatible agent's input, and when types are compatible but not identical, Weave auto-injects a transform.

from weaveflow import agent, DataType, Pipeline

@agent(name="summarizer", input=DataType.TEXT, output=DataType.TEXT,
       tags=["summarization"], llm="anthropic:claude-opus-4-8")
async def summarize(ctx):
    return await ctx.complete(f"Summarize:\n{ctx.input.value}")

result = await Pipeline([summarize]).run("a long document ...")

Why Weave

Problem today	Weave
Agents are locked to one framework	Open port contract; any compliant agent connects
LLM vendor lock-in	Swap brains via a "provider:model" string
Custom glue code between agents	Connection protocol validates + auto-transforms handoffs
Hard to test multi-agent chains	In-process LocalRunner with per-hop tracing

Install

make install                       # dev + all providers in a venv
# or, minimal core (zero deps) + one provider:
pip install "weaveflow[anthropic]"

The core has no runtime dependencies. Provider SDKs are optional extras: weaveflow[openai], weaveflow[anthropic], weaveflow[google], weaveflow[mistral], weaveflow[ollama], weaveflow[deepseek], weaveflow[all].

Core concepts

Weave models an agent as a body with a standard anatomy (PRD §5.3):

Layer	Weave component
Skin (ports)	PortSchema, the input/output typed contracts
Brain (LLM)	LLMAdapter, swappable via create_adapter("provider:model")
Memory	ShortTermMemory, LongTermMemory (vector search)
Nervous system	Guardrails: pre / post / on_error hooks
Connection	ConnectionProtocol + Router + transforms

Standard data types

text, structured_json, image, code, audio, document, embedding, stream.

Two ways to define an agent

# 1. Decorator (ergonomic)
@agent(name="x", input=DataType.TEXT, output=DataType.TEXT, llm="openai:gpt-4o")
async def x(ctx): ...

# 2. Subclass BaseAgent (full control / DI)
class MyAgent(BaseAgent):
    async def handle(self, ctx): ...

Swap LLM backends

"openai:gpt-4o" · "anthropic:claude-opus-4-8" · "google:gemini-1.5-pro"
"mistral:mistral-large-latest" · "deepseek:deepseek-chat" · "ollama:llama3"

Set the matching API key env var (OPENAI_API_KEY, ANTHROPIC_API_KEY, …). Ollama runs locally and needs none.

Compose and connect

from weaveflow import Pipeline, Parallel, LocalRunner

# Series:
pipe = Pipeline([cleaner, extractor, summarizer], llm="anthropic:claude-opus-4-8")
out = await pipe.run("raw input")

# Fan-out / fan-in (runs branches concurrently, then merges):
fan = Parallel([sentiment, topics, summary], input=DataType.TEXT)

# Any variant: Parallel is a BaseAgent, so it nests inside a Pipeline:
pipe = Pipeline([cleaner, Parallel([analyze_a, analyze_b]), report])

trace = await LocalRunner().simulate([cleaner, extractor], "raw input")
for hop in trace.hops:
    print(hop.agent, hop.output.value, hop.elapsed_ms)

Connect a foreign agent, no rewrite

If you can call it from Python, you can connect it to Weave. Already have an agent in LangChain, LangGraph, or CrewAI? Wrap it and plug it in, and the handoff auto-calibrates. Anything else (a plain function, a bound method, an HTTP or SDK call) goes through from_callable, the universal escape hatch. The bridges are duck-typed, so the zero-dependency core stays clean, and a wrapped agent becomes a BaseAgent that composes in Pipeline and Parallel exactly like a native one.

from weaveflow import from_langchain, from_crewai, from_callable, Pipeline

theirs = from_langchain(their_langchain_chain)   # also works with LangGraph graphs
mine   = my_weave_agent
out = await Pipeline([theirs, mine]).run("...")   # connected, no rebuild

# or any function / SDK call:
agent = from_callable(lambda text: external_sdk.run(text), name="legacy")

See docs/guide-interop.md.

CLI

weaveflow scaffold my-agent           # create a starter agent file
weaveflow validate my_agent.py        # validate ports + print manifest
weaveflow package my_agent.py         # portable .weave.zip (code + manifest.json)

Project layout

weaveflow/
  types/        # 8 primitive data types + compatibility graph
  schema/       # port schemas, validator, registry
  llm/          # adapter abstraction + 6 providers + factory
  memory/       # short-term buffer + long-term vector store
  guardrails/   # pre/post/on-error hooks
  agent/        # BaseAgent, context, @agent decorator
  connection/   # protocol engine, router, transforms
  interop/      # bridges: from_langchain / from_crewai / from_callable
  runtime/      # Pipeline (series) + Parallel (fan-out/in) + LocalRunner
  cli/          # scaffold / validate / package
tests/          # mirrors weaveflow/ one-to-one
examples/       # offline quickstart
example-agents/ # copy-paste agent templates (one per folder) + a connect demo

Development

make check      # lint + typecheck + tests (≥80% coverage gate)
make example    # run the offline quickstart
make docker-build

License

Apache-2.0.