CogSynth is a Python SDK for composing agents into a graph — with typed tool-calling, a shared context ledger, and an evaluation harness you run before you ship.
No API keys, no vendor lock-in. Bring your own model.
$ pip install cogsynth
from cogsynth import Agent, Graph, Tool
graph = (
Graph("research")
.add(Agent("planner", tools=[web_search]))
.add(Agent("writer"))
.edge("planner", "writer")
)
result = graph.run("Summarize the v0.14 changelog.")
print(result.output)CogSynth gives you five primitives. Compose them; the runtime handles scheduling, retries, and state.
Agents are nodes in a graph. Edges define control flow and data flow, so a workflow reads like a wiring diagram instead of a chat transcript.
Give an agent typed tools with JSON schemas. Arguments are validated at the boundary, not pasted into a prompt and hoped for.
One append-only store for shared state across agents. No hidden prompt stuffing, no recomputing what a sibling already produced.
Run a graph against fixtures and judges. A change that moves a score fails CI before it reaches a user.
Execution is a directed acyclic graph. Fan-out and join points are explicit, so concurrency is a property of the graph, not of your luck.
Every node write lands in the ledger. Interrupt a run and resume it without replaying side effects.
Build a report pipeline that plans, researches, and writes. Everything below runs with a stock Python 3.11+ interpreter and your own model endpoint.
Install the package and pin a model. CogSynth is provider-agnostic: point it at whatever endpoint you already pay for.
$ pip install cogsynth
$ export COGSYNTH_MODEL="anthropic/claude-sonnet-4"
$ export ANTHROPIC_API_KEY="sk-..."Tools are ordinary functions. The decorator derives a JSON schema from your type hints, so the model sees a strict contract instead of a prose description.
from cogsynth import Tool
@Tool
def web_search(query: str, n: int = 5) -> list[dict]:
"""Search the web and return result snippets."""
return client.search(query, limit=n) # your HTTP client hereWire the graph. Each agent declares its role, its tools, and which node it hands off to next.
from cogsynth import Agent, Graph
from cogsynth.ledger import ContextLedger
from tools import web_search
ledger = ContextLedger(name="report-builder")
graph = (
Graph("report-builder", ledger=ledger)
.add(Agent("planner", instruction="Decompose the task into queries.", tools=[web_search]))
.add(Agent("researcher", instruction="Run the queries and record findings.", tools=[web_search]))
.add(Agent("writer", instruction="Turn findings into a tight brief."))
.edge("planner", "researcher")
.edge("researcher", "writer")
)Run it and read the result. RunResult exposes the terminal output, the full context ledger, and per-node timings.
$ python workflow.py
# planner → 2 queries 1.2s
# researcher → 4 tool calls 2.9s
# writer → 1 draft 1.0s
result.output # → "The v0.14 release adds checkpoint/resume..."Five public classes cover the whole SDK. Signatures are stable across 0.x; breaking changes are announced in the changelog.
A single unit of work in a graph. An agent runs its instruction against the context ledger, may call its tools, and returns a message that downstream nodes read.
| param | type | description |
|---|---|---|
name | str | Unique node identifier within the graph. |
instruction | str | The system prompt for this node. Written once, evaluated on every run. |
tools | list[Tool] | Tools this agent may invoke. Defaults to none. |
model | str | None | Override the graph-level model for this node only. |
temperature | float | Sampling temperature. Defaults to 0.0 for deterministic runs. |
The orchestration DAG. Methods chain so you can build the topology in one expression, or mutate it incrementally.
| method | signature | description |
|---|---|---|
add | add(agent: Agent) → Graph | Register a node. Returns self for chaining. |
edge | edge(src: str, dst: str | list[str]) → Graph | Add a directed edge. A list of destinations creates a fan-out. |
run | run(input: str, *, stream=False) → RunResult | Topologically execute the graph and return the terminal result. |
resume | resume(run_id: str) → RunResult | Continue an interrupted run from its last checkpoint. |
Wraps a plain function into a model-callable tool. The JSON schema is derived from annotations and the docstring; arguments are validated before the function runs, and failures are returned to the agent as structured errors rather than thrown into the transcript.
An append-only, queryable store of everything a run has produced. Agents read from and write to it through the SDK; you never assemble the shared context by hand.
| method | signature | description |
|---|---|---|
append | append(kind: str, body: dict, *, node: str) → Entry | Write a typed entry (fact, finding, draft, …). |
query | query(kind: str | None = None, node: str | None = None) → list[Entry] | Read entries back, filtered by type or producer. |
checkpoint | checkpoint() → str | Persist the ledger and return a run id usable by Graph.resume. |
Replays a graph against fixture inputs and scores each output with deterministic judges. Use it in CI; a regression in a scored behavior fails the run before the change ships.
| method | signature | description |
|---|---|---|
evaluate | evaluate(graph: Graph, fixtures: list[Fixture]) → Report | Run every fixture through the graph and score the outputs. |
Each pattern below is a topology, not a prompt. Swap the model, keep the edges.
Plan → search → synthesize. The planner emits queries, the researcher records findings in the ledger, the writer composes the brief. Deterministic handoff at every edge.
Route the ticket, call a policy tool, draft the reply. A fan-out edge lets the router split to lookup and policy in parallel, then rejoin at the drafter.
One node reviews the diff against repo rules; another regenerates the docstring contract. Both read the same ledger, so the doc change always matches the reviewed code.
Extract typed records from messy inputs with strict tool schemas. The harness scores field accuracy against fixtures, so you tune the prompt without regressing the parser.
Design decisions and gotchas from developing an orchestration runtime.
Why we model agent coordination as a graph with explicit edges instead of letting a loop talk to itself until a token budget runs out.
Cycles are where state leaks. Enforcing a DAG at build time is the cheapest concurrency bug you will ever prevent.
How a typed, append-only store keeps shared context observable instead of silently embedded in a growing system message.
Prompt changes need CI too. A harness that replays fixtures and scores outputs turns “it feels better” into a pass/fail.
Derive the schema from your type hints, not from the model’s native format. Your tools should outlive the model you happen to be using.
Follows Keep a Changelog. Breaking changes land in minor releases during 0.x and are called out explicitly.
Graph.resume() and ContextLedger.checkpoint() for interrupt-and-resume runs.run(stream=True) emits per-node progress events.RunResult.output is now a str, not a Message.EvaluationHarness gains first-class Judge objects (regex, exact-match, and model-as-judge).ContextLedger.query() accepts kind and node filters.edge(src, [a, b]).Agent, Graph, Tool, and ContextLedger.“The context ledger is the part we actually trust. We stopped concatenating state into prompts and our review pass got faster.”
“Swapping the model underneath a workflow is a one-line change now. Our tools keep their schemas and the eval harness catches the rest.”
A small team building a deterministic runtime for multi-agent Python. Six of us, one graph at a time.
Previously led orchestration infrastructure at a large cloud provider. Sets the roadmap and keeps the SDK honest about what “deterministic” means.
Distributed systems and LLM tooling. Designed the graph runtime and the append-only context ledger that every run writes through.
Owns scheduling, retries, and checkpoint/resume. Obsessed with making fan-out and joins behave the same on a laptop and in CI.
Runs the docs, the quickstart, and the API surface. If a workflow takes more than five minutes to explain, it gets rewritten.
Builds the typed tool-calling layer and the provider integrations. Makes schemas survive a model swap.
Owns the evaluation harness and the judge library. A regression that moves a score never reaches a user on her watch.