Bind the participants, then run the coding session.
The protocol and filesystem service are ready. The remaining code binds three model-backed agents and an interactive terminal client, then starts the session.
Continue appending each block to coding_agent.py.
The protocol constrains message order and available labels. Briefs supply the judgement required within those bounds: what to inspect, how to report a patch, and when a review issue deserves the single correction.
PLANNER_BRIEF = (
'Explore with List and Read only. List the workspace, read the files '
'you need, then Ready and send Plan: what to change and in which file. '
'Do not invent files you have not read.'
)
CODER_BRIEF = (
'Carry out Plan. Write the full file contents. When the change is in '
'place, Done and send Patch describing what changed. After Ship, send '
'Idle then Delivered. After Fix, Write once more, then Delivered.'
)
REVIEWER_BRIEF = (
'Review Patch against Task and Plan. Prefer Ship when the change '
'matches the task. If something concrete is missing, send Fix in at '
'most twenty words — you review once.'
)Repair(attempts=2) below permits two bounded retries when a model response
cannot be decoded as a currently available protocol choice.
Create one OpenAI client, then attach a participant to every role. The planner, coder, and reviewer share the model while retaining separate briefs and local protocol states. The workspace receives deterministic tools; the client uses interactive terminal I/O.
def _model() -> ap.OpenAIModel:
client = AsyncOpenAI(
api_key=os.environ['OPENAI_API_KEY'], max_retries=0, timeout=30.0
)
return ap.OpenAIModel(MODEL, client)
def build_cast(
root: Path,
client_io: Any | None = None,
model: Any | None = None,
) -> ap.Cast:
model = _model() if model is None else model
client_io = ap.CliHumanIo() if client_io is None else client_io
return (
ap
.Cast(protocol)
.play(Client, ap.human(client_io))
.play(
Planner,
ap.agent(model, PLANNER_BRIEF, repair=ap.Repair(attempts=2)),
)
.play(Coder, ap.agent(model, CODER_BRIEF, repair=ap.Repair(attempts=2)))
.play(
Reviewer,
ap.agent(model, REVIEWER_BRIEF, repair=ap.Repair(attempts=2)),
)
.play(Workspace, ap.service(*workspace_tools(root)))
)The optional client_io and model parameters make the cast usable with
deterministic participants in tests. A normal run constructs the CLI and OpenAI
participants automatically.
Finish the script with its entry point:
def main() -> None:
project_all(protocol)
report = debug.Report()
report.protocol(protocol)
root = Path.cwd()
print(f'workspace: {root}')
print('Enter the task when Client prompts; the protocol drives the rest.')
trace = build_cast(root).run_sync(allowance=ap.Allowance(unfoldings=12))
report.conversation(trace)
if __name__ == '__main__':
main()project_all checks that the global protocol yields a valid endpoint for every
role. Allowance(unfoldings=12) bounds recursive planning and implementation
steps. debug.Report prints both the declared protocol and the completed trace.
Export your API key, change to the directory the agent may edit, and pass the
script's path to uv run:
export OPENAI_API_KEY=...
cd path/to/your/project
uv run /absolute/path/to/coding_agent.pyAt the client prompt, select Task and enter a concrete request, for example:
Add a function named matrix_multiply to matrix.py and cover incompatible dimensions.The planner will inspect the directory before publishing its plan. The coder
writes complete file contents, the reviewer makes one decision, and the final
trace ends with Delivered. The working directory is the workspace root, so run
the command inside a disposable repository while experimenting.
The completed harness keeps its extension points visible. Change the protocol
when authority or control flow has to change, and change a brief to sharpen a
decision inside that flow. Swapping the workspace service points the whole thing
at a different execution environment. See
participants and roles for participant
semantics and protocol-first design for the
projection model used by Cast.