I built this — reactifact, a Python agent runtime that also speaks MCP natively, both as a client and a server. Here's the argument for why it exists.
A knowledge-agent question like "why did infra costs jump in Q2?" usually needs Confluence and GitLab and a CSV calculation and, sometimes, a human to confirm a number before it ships. The next question needs a different subset of those. Multiply that by a real product surface and you're not writing an agent anymore — you're maintaining a graph of add_edge/ add_conditional_edges calls that has to be re-drawn every time the shape of a question changes.
That's not a LangGraph problem specifically — it's what happens whenever the orchestration is the code. You're modeling every path a question could take, by hand, up front.
reactifact flips which thing you write down. You don't draw a path from A to B. You declare, per agent, what it consumes and what it produces — typed artifacts, not string blobs in a shared dict. The runtime watches what actually exists and runs whichever agent's consumes just got satisfied. Two agents that have never heard of each other compose correctly as long as one produces what the other needs.
Here's the whole thing, runs offline, no API key:
from pydantic import BaseModel
from reactifact import Budget, Consume, Context, Runtime, RuntimeResources, create_agent, produce
class Question(BaseModel):
text: str
class Evidence(BaseModel):
text: str
class Answer(BaseModel):
text: str
DOCS = {
"refund": "Refunds are available within 14 days of purchase.",
"pricing": "The Pro plan is $49/month, billed annually.",
}
@produce(Evidence)
async def find_evidence(context, inputs, event, effects):
question = next((a for a in inputs if isinstance(a.data, Question)), None)
if question is None:
return None
hit = next((v for k, v in DOCS.items() if k in question.data.text.lower()), None)
if hit is not None:
effects.create(Evidence(text=hit))
@produce(Answer)
async def answer_from_evidence(context, inputs, event, effects):
evidence = next((a for a in inputs if isinstance(a.data, Evidence)), None)
if evidence is None:
return None
effects.create(Answer(text=evidence.data.text)).link("supported_by", evidence)
search_agent = create_agent("search", consumes=[Consume(Question)], produces=[find_evidence])
answer_agent = create_agent("answer", consumes=[Consume(Evidence)], produces=[answer_from_evidence])
ctx = Context(resources=RuntimeResources())
runtime = Runtime(ctx, agents=[search_agent, answer_agent], budget=Budget(max_runs=10))
ctx.create(Question(text="what's your refund policy?"))
runtime.run() # search_agent and answer_agent both react — nobody wired them together
answer = ctx.latest(Answer)
evidence = ctx.related(answer.id, "supported_by")[0]
print(answer.data.text) # "Refunds are available within 14 days of purchase."
print("supported_by:", evidence.data.text) # provenance you can trace, not just a string in a log
No edge between search_agent and answer_agent exists anywhere in this file. answer_agent fires the instant an Evidence artifact lands in Context — because it declared consumes=[Consume(Evidence)], not because anyone told it "run after search." Add a third agent that also produces Evidence from a different source next month, and answer_agent still fires, unmodified.
State is typed and versioned, not a dict. Every artifact is a pydantic model with an id, a version, and history. context.diff(v1, v2) is a real operation — not something you reconstruct from logs after the fact.
Provenance is built in, not bolted on. That .link("supported_by", evidence) call above isn't a debugging add-on — it's a real edge the runtime stores. Answer —supported_by→ Evidence —extracted_from→ Doc is queryable. "Why did the agent say that?" has an actual answer instead of a grep through message history.
Calculations are calculated. A recipe pushes arithmetic into a deterministic code path, not the model's next-token guess. In the demo below, "$3,580" comes from sum() over a CSV column, and the answer says so — not "approximately."
Short version, next to the two frameworks people usually compare this to:
| LangGraph | CrewAI | reactifact | |
|---|---|---|---|
| Primary abstraction | explicit state graph (nodes + edges) | role-based crew of agents | typed artifacts + reactive agents |
| Control flow | you draw it | mostly fixed (sequential/hierarchical) | derived from state changes |
| State | a shared, loosely-typed dict/ TypedDict |
task outputs passed along | versioned, typed, immutable-per-version artifacts |
| "Why did it say that?" | manual logging/checkpoint inspection | not tracked by default | provenance graph ( supported_by /derived_from /…) built in |
| Numbers/calculations | the LLM computes unless you write a tool | same | recipes push calculation into deterministic code |
| Rollback / branching | checkpointer + manual replay logic | not built in | context.branch() , three-waymerge() , deterministic replay |
| MCP | via langchain-mcp-adapters (client) |
via MCPServerAdapter (client) |
client and server, built in |
| Maturity / ecosystem | high, widely used in production | high, large community | pre-1.0, one maintainer |
Full version, written as a comparison and not a pitch — including where reactifact is the wrong call — in docs/en/comparison.md.
Here's the CLI from the knowledge example answering a question that touches docs and a spreadsheet:
And the shape of what's actually happening — two independent agent groups, neither aware of the other, both required before the answer fires:
reactifact.mcp (an optional extra — the core has no dependency on it) goes
both ways:
mcp_stdio_tools/ mcp_http_tools connect to any MCP server and
hand back its tools as ordinary Tool s — usable by ToolUse/ LLMAgent
exactly like a local @tool function, no separate code path.create_mcp_server(tools, context=ctx) exposes reactifact's
own **kwargs — and, with context=, a running Context's artifacts
as two read-only resources. Claude Desktop, Claude Code, or another agent
can call straight into a live reactifact app.
from reactifact import Consume, create_agent
from reactifact.mcp import mcp_stdio_tools
from reactifact.tool_use import ToolUse
async with mcp_stdio_tools("npx", ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]) as tools:
fs_agent = create_agent("fs", consumes=[Consume(Question)], produces=[
ToolUse("Answer questions about files in /tmp.", tools),
])
effects.ask(...) → PendingQuestion, and resumes on the next message instead of restarting. No special "interrupt" plumbing.Context.branch() + three-way merge()
I built this alone, it's pre-1.0, there's no funding and no managed platform
behind it. I'd rather say that here than have you find out after adopting it —
along with the more specific cases where it's the wrong call:
reactifact's Source abstraction is intentionally small — filesystem, CSV, embeddings, web — you write the rest (MCP narrows this specifically for tool-calling, not for retrieval).
pip install reactifact
If you've hit the "the next question needs a different graph" wall, I'd genuinely like to know whether this model holds up outside my own use case — issues and PRs both welcome.