Bring your own agent
For: plugin authors.
Wrap an existing Python agent, LangGraph graph or CrewAI crew with the byo-agent template. The template is a harness: one process per job that exchanges JSON lines over stdin and stdout. Your code keeps its agent loop.
The platform supplies the durable job record, live stream and trace. Model calls pass through the gateway for metering and budgets, with the tenant key kept on the edge. The harness asks before each tool call so hooks, default-deny policy and approvals can decide it.
An agent on Microsoft Agent Framework, Google ADK or AWS Strands has a template of its own, with the framework's model client already pointed at the gateway and its tool calls already asked: see Frameworks as harnesses.
Scaffold it
oz0 plugin new acme-triage --template byo-agent --description "Sorts support tickets"
cd acme-triage
uv lock && uv sync
uv run pytest
oz0 plugin lint
acme-triage/
├── oz0-plugin.yaml a harness acme-triage, and the agent
├── agents/acme-triage.yaml runtime: harness:acme-triage, its harness.tools
├── prompts/acme-triage.md the instructions, the system prompt of its model calls
├── src/acme_triage/
│ ├── harness.py the protocol: session request in, events out, permission requests
│ ├── agent.py run(session): your agent
│ ├── legacy.py an existing script, the worked example
│ └── adapters.py LangGraph and CrewAI
├── tests/test_harness.py the harness over a pipe with a fake model API
├── AGENTS.md
└── .claude/skills/oz0-plugin/SKILL.md
apiVersion: oz0/v1
name: acme-triage
version: 0.1.0
description: "Sorts support tickets"
runtime: python
harnesses:
- name: acme-triage
command: ["python", "-m", "acme_triage"]
permissions: true # it asks before each tool call
tools: [notify_team] # what a session may call when its agent lists no harness.tools
formats: [anthropic-messages] # the model API your code speaks
agents: [agents/]
skills: [skills/]
name: acme-triage
description: "Sorts support tickets"
runtime: harness:acme-triage
model:
use: default # an alias, resolved to a record that takes the harness's formats
instructions: prompts/acme-triage.md
input: text
output: text
limits:
max_steps: 10 # the most model calls a session makes
timeout: 10m
harness:
tools: [notify_team]
# requires_approval: [notify_team]
Your agent is run(session)
agent.py is where your code goes. It gets a Session and returns the answer, text or anything JSON can hold:
from .harness import Refused, Session
from .legacy import triage
def notify_team(team: str, summary: str) -> str:
"""Tell a team about a ticket."""
return f"told the {team} team: {summary}"
def run(session: Session) -> str:
session.say("Sorting the ticket. ")
result = triage(session.text, complete=session.complete) # your existing code, unchanged
try:
session.call_tool("notify_team", {"team": result["team"], "summary": session.text[:200]}, notify_team)
except Refused as why:
session.say(f"(The team was not told: {why}) ")
return f"{result['team']}: {result['reply']}"
session. | What it does |
|---|---|
text, input | The job's input, as text, or as it came (JSON for an agent with an input schema) |
say(text) | Streams text to the job's live view |
complete(prompt) | One model call through the node's LLM proxy (Anthropic's Messages API), with the agent's instructions as the system prompt; counts toward limits.max_steps |
call_tool(name, arguments, fn) | Asks the job, then runs fn(**arguments) with the arguments it allowed, and shows the call and its result in the job; raises Refused with the reason when the job says no |
job, agent, model, workdir, state_dir, options | The request's fields: the agent definition's harness: options are in options |
legacy.py, the worked example, is a script that never heard of Orchestrator Zero: it takes a complete function and returns a team and a reply. Anything that calls a model can be wrapped like that.
Model calls
The session's process gets ANTHROPIC_BASE_URL and ANTHROPIC_API_KEY (and the OPENAI_ pair) pointing at the node's LLM proxy, with a key of the session's own: never the tenant's provider key. session.complete() uses them, and so does any SDK that reads those variables, Anthropic's and OpenAI's among them, LangChain's and CrewAI's through them. Every call is metered under the job, and the job's budget, the tenant's quotas and kill switches apply.
The agent's model: {use: default} resolves to one of the tenant's provider records that takes a format the harness lists in formats (Anthropic's Messages API in the template); list OpenAI's too if your code speaks it. complete() also sends the job and agent as the Oz0-Job and Oz0-Agent headers, which the harness protocol asks for.
Tool calls are asked first
The agent definition lists the tools its code may call under harness.tools. Before each call, session.call_tool() sends a permission request to the runtime and waits:
- A tool the agent does not list is refused at once.
- The tenant's hooks see it as
<harness>.<tool>, hereacme-triage.notify_team: they allow it, deny it, change its arguments or ask a person. - A tool in
harness.requires_approvalwaits for a person's approval, in the inbox, for as long as it takes.
The answer comes back on stdin; only an allowed call runs, with the arguments the job allowed. The call and its result show in the job's live stream and trace like any tool call.
LangGraph and CrewAI
adapters.py runs a LangGraph graph or a CrewAI crew as run(session), and asked() turns any function into a tool the job decides. Add the framework to pyproject.toml, pinned, and call them from agent.py:
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from .adapters import asked, run_langgraph
def run(session):
lookup = tool(asked(session, "lookup_order", lookup_order)) # asked before every call
graph = create_react_agent(ChatAnthropic(model=session.model_name), [lookup], prompt=session.instructions)
return run_langgraph(session, graph)
from crewai import LLM, Agent, Crew, Task
from crewai.tools import tool
from .adapters import asked, run_crew
def run(session):
lookup = tool("lookup_order")(asked(session, "lookup_order", lookup_order))
agent = Agent(role="Support", goal=session.instructions, backstory="", llm=LLM(model=session.model), tools=[lookup])
task = Task(description="{input}", expected_output="A reply to the customer", agent=agent)
return run_crew(session, Crew(agents=[agent], tasks=[task]))
List each tool (lookup_order) under harness.tools. A refused call becomes the tool's answer (refused: <why>), so the framework's model reads why and goes on. Pin langgraph and langchain-anthropic, or crewai[anthropic]: never CrewAI's litellm extra, which Orchestrator Zero keeps out of everything it runs after the LiteLLM supply-chain incident of March 2026.
Test it and run it
uv run pytest # the harness over a pipe, as a node runs it, with a fake model API
oz0 dev # a dev server and a node with the plugin
In a second terminal, with the export OZ0_CONTEXT=... line dev printed and the tenant's key set:
printf %s "$ANTHROPIC_API_KEY" | oz0 secret set ANTHROPIC_API_KEY
oz0 run acme-triage "I was charged twice for my order"
acme-triage: Sorting the ticket.
acme-triage → notify_team {"summary":"I was charged twice for my order","team":"billing"}
acme-triage ← notify_team told the billing team: I was charged twice for my order
billing: Sorry about the double charge; we have refunded the second one.
Uncomment requires_approval: [notify_team] in the agent, and the next run waits for a person in oz0 approval list before it tells the team. scripts/e2e-kit.sh in the repository runs all of this against a fake model API.
What a session is
One process per job: the node starts python -m acme_triage in the plugin's environment, writes the session request to its stdin and reads its events, as for every harness. It gets an empty working directory and a private state directory, removed when the session ends, a minimal environment, and the sandbox of its node's tier. Cancelling the job sends SIGTERM to its process group. A session that fails fails the job, and is not retried.