weaveflow 2.1.0

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

Weaveflow: 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.

Weaveflow 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, Weaveflow 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 Weaveflow

Problem today	Weaveflow
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

# minimal core (zero runtime 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].

Quickstart

from weaveflow import agent, DataType, Pipeline, Parallel, LocalRunner

# Define an agent: a decorator (ergonomic) or a BaseAgent subclass (full control).
@agent(name="x", input=DataType.TEXT, output=DataType.TEXT, llm="openai:gpt-4o")
async def x(ctx): ...

# Compose in 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). Parallel is itself a
# BaseAgent, so it nests inside a Pipeline:
pipe = Pipeline([cleaner, Parallel([analyze_a, analyze_b]), report])

# Trace every hop in-process, no network:
trace = await LocalRunner().simulate([cleaner, extractor], "raw input")
for hop in trace.hops:
    print(hop.agent, hop.output.value, hop.elapsed_ms)

Standard data types

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

Swap LLM backends

Pass any "provider:model" string; set the matching API key env var (OPENAI_API_KEY, ANTHROPIC_API_KEY, …). Ollama runs locally and needs none.

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

Connect a foreign agent, no rewrite

If you can call it from Python, you can connect it to Weaveflow. 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.

from weaveflow import from_langchain, from_crewai, from_callable, Pipeline

theirs = from_langchain(their_langchain_chain)   # also works with LangGraph graphs
out = await Pipeline([theirs, my_weaveflow_agent]).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 .weaveflow.zip (code + manifest.json)

Documentation

Full guides — agents, LLM backends, memory, guardrails, connections, and interop — live in docs/. Runnable templates are in example-agents/.

License

Apache-2.0.