Production-readiness¶
The boring stuff that turns a multi-agent demo into something you'd deploy. Every primitive on this page works inside any of the eight workflow shapes — you don't pick "shape" or "production-ready", you get both.
Typed terminal artifacts¶
from pydantic import BaseModel
from tulip.agent import Agent, AgentConfig
class Postmortem(BaseModel):
severity: str
root_cause: str
timeline: list[str]
action_items: list[str]
writer = Agent(config=AgentConfig(
model="openai:gpt-4o-mini",
output_schema=Postmortem,
))
result = writer.run_sync("Write a postmortem for incident #4421")
postmortem: Postmortem = result.parsed # validated, not free text
output_schema validates the model's final answer against a Pydantic
schema. The workflow's terminal node returns a typed object — Verdict,
Postmortem, ConcessionDecision, ContractDecision — that the rest of
your system can consume without a brittle JSON re-parse.
Used by notebooks 32 (debate), 63 (incident), 64 (concession approval), 65 (DPA review).
→ See Structured output.
Idempotent tools — deduplicate identical calls¶
from tulip.tools import tool
@tool(idempotent=True)
def isolate_host(host_id: str, case_id: str) -> dict:
return edr.quarantine(host_id, case_id)
The ReAct loop reuses the recorded result for identical (name, kwargs) calls
in its documented run/checkpoint scope. A downstream idempotency key is still
required to cover the window where an external action succeeds before its
receipt is durably recorded.
→ See Idempotency.
Durable memory — survive every restart¶
from tulip.agent import Agent, AgentConfig
from tulip.memory.backends.s3 import S3Backend
agent = Agent(config=AgentConfig(
model="openai:gpt-4o-mini",
checkpointer=S3Backend(bucket="tulip-state", prefix="..."),
))
Eight backend implementations, one Checkpointer Protocol — S3,
Postgres, MySQL, Redis, OpenSearch, HTTP, file, and in-memory. The
graph snapshots state at every interrupt() boundary; you can pause
for a human Friday afternoon and resume Monday morning from a
different process, region, or runtime.
→ See Checkpointers.
Reflexion — catch a bad turn before the next one¶
reflexion=True self-evaluates every turn and feeds the next Think a
sharper plan. When a critic loop is overkill or you want intra-agent
self-correction, flip the flag.
→ See Reasoning.
Grounding — verify claims against their source¶
The configured grounding judge scores the final answer against the tool
evidence collected during the run; if the score is below the threshold
(0.65 by default), the loop replans, up to max_replans times (2 by
default). For typed grounding (entity-level evidence and
attribution), use GSAR.
Streaming events — every node visible¶
from tulip.core.events import ToolStartEvent, TerminateEvent
from tulip.multiagent import StreamMode
async for event in graph.stream(initial, mode=StreamMode.NODES):
match event:
case ToolStartEvent(tool_name=n, agent_name=a):
print(f"{a} → {n}")
case TerminateEvent(final_message=m, agent_name=a):
print(f"{a} done: {m}")
Every shape in the framework emits the same typed event taxonomy.
agent_name is set on every event, so you can attribute output back to
the specialist that produced it. SSE-ready, match-statement friendly,
identical shape whether the back-end is a single agent, an
orchestrator, a swarm, or an A2A mesh.
→ See Streaming · Events · Graph streaming.
Observability — traces, metrics, hooks¶
from tulip.agent import Agent, AgentConfig
from tulip.hooks.builtin import TelemetryHook
agent = Agent(config=AgentConfig(
model="openai:gpt-4o-mini",
hooks=[TelemetryHook(service_name="tulip-incident-bot")],
))
OpenTelemetry wired through every event. Hooks let you observe and
steer each model and tool call (BeforeModelCallEvent,
AfterModelCallEvent, BeforeToolCallEvent, AfterToolCallEvent)
and wrap the whole run with the on_before_invocation /
on_after_invocation methods, without touching the graph.
→ See Observability · Hooks.
Safety & guardrails¶
from tulip.agent import Agent, AgentConfig
from tulip.hooks.builtin.guardrails import GuardrailsHook, GuardrailConfig
agent = Agent(config=AgentConfig(
model="openai:gpt-4o-mini",
hooks=[GuardrailsHook(config=GuardrailConfig())],
))
Input validation, PII redaction, topic policies, and tool restrictions
ride on the same hook system, so nothing is bolted-on. Stack them
freely. One caveat: hooks attach per-agent, so in orchestrated shapes
gate the side-effecting tool itself with
admit() — the gate travels with the tool,
and no orchestration path bypasses it.
→ See Safety & Guardrails.
Evaluation¶
from tulip.evaluation import EvalCase, EvalRunner
cases = [
EvalCase(name="terminates", prompt="...", expected_output_contains=["resolved"]),
EvalCase(name="uses_siem", prompt="...", expected_tools=["query_siem"]),
]
report = EvalRunner(agent=graph).run(cases)
Run regression suites against any agent or graph. Failures point at the specific node and event that diverged.
→ See Evaluation.
Putting it together¶
A notebook-63-style incident-response graph in production looks like:
from tulip.agent import Agent, AgentConfig
from tulip.multiagent.graph import StateGraph, GraphConfig
from tulip.memory.backends.s3 import S3Backend
graph = StateGraph(config=GraphConfig(
allow_cycles=True,
max_iterations=20,
checkpointer=S3Backend(bucket="incidents", prefix="..."),
))
# ... nodes use Send for parallel investigation, interrupt() for the
# severity gate, output_schema=Postmortem for the terminal artifact,
# idempotent tools for paging, hooks for OTel spans.
Pick a shape directly, or let the loop reach for a shape tool and build the right one from a typed intent. Then wire the primitives above through it and ship it.