diff --git a/examples/coding-agent/README.md b/examples/coding-agent/README.md
new file mode 100644
index 000000000..cdb94b310
--- /dev/null
+++ b/examples/coding-agent/README.md
@@ -0,0 +1,93 @@
+
+
+# Coding agent
+
+A minimal coding agent harness: an LLM that reads files, writes files, and runs
+shell commands in a loop until it has finished a task.
+
+The point of this example is the *loop*. A tool-calling assistant runs one tool
+and returns to the user. An agent keeps going on its own -- it reads a file,
+sees what's in it, edits it, runs the tests, sees them fail, and tries again --
+without a human in between. That is a cycle, and expressing cycles clearly is
+what Burr is for.
+
+
+
+## The loop
+
+| Action | What it does |
+| --- | --- |
+| `human_input` | Takes the task and seeds the message history |
+| `create_prompt` | A seam for shaping the prompt before each call -- add a file tree, a summary of earlier steps, retrieved context |
+| `call_llm` | Asks the model what to do next: call a tool, or finish |
+| `read_file` / `write_file` / `list_files` / `run_bash` | One bound action per tool, so each shows up separately in the Burr UI |
+| `respond` | Surfaces the answer, or reports that the step budget ran out |
+
+Two transitions leave `call_llm` for `respond`: one when the model answers
+without requesting a tool, and one when `steps` reaches `max_steps`. Both exits
+are listed before the tool transitions, because Burr takes the first condition
+that matches. An agent that decides its own next step can loop forever, so the
+budget is not optional.
+
+Tool results are appended to `messages` as `role: "tool"` entries carrying the
+`tool_call_id` they answer. That is what lets the model see what its last action
+actually returned.
+
+## Running it
+
+ pip install -r requirements.txt
+ python application.py
+
+With no API key set, the example uses a scripted client that replays a fixed
+sequence of tool calls, so it runs immediately and in CI. Set `OPENAI_API_KEY`
+to use a real model instead.
+
+Regenerating `statemachine.png` needs the Graphviz binary (`apt-get install
+graphviz`), not just the Python package.
+
+To watch a run in the Burr UI:
+
+ burr
+
+## Safety
+
+`run_bash` executes shell commands with the permissions of the process running
+the agent. File paths are confined to the workspace directory, but a shell
+command can simply `cd` out of it. **This is a teaching example, not a sandbox.**
+Run it against a directory you do not mind losing, ideally inside a container.
+
+Adding a permission step before tool execution -- prompting for approval, or
+checking an allowlist -- is the natural next thing to build.
+
+## Known simplifications
+
+- One tool call is executed per iteration; if the model requests several, the
+ extras are dropped.
+- `create_prompt` is a passthrough. It exists to show where prompt construction
+ belongs, not because it does anything yet.
+- Message history grows without bound. A longer-running agent needs summarisation
+ or truncation.
+
+## Files
+
+- [application.py](application.py) -- the state machine, actions, and LLM clients
+- [tools.py](tools.py) -- the four tools
+- [notebook.ipynb](notebook.ipynb) -- the same example, walked through step by step
+- [requirements.txt](requirements.txt) -- the environment
diff --git a/examples/coding-agent/__init__.py b/examples/coding-agent/__init__.py
new file mode 100644
index 000000000..13a83393a
--- /dev/null
+++ b/examples/coding-agent/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git a/examples/coding-agent/application.py b/examples/coding-agent/application.py
new file mode 100644
index 000000000..d1ab87be8
--- /dev/null
+++ b/examples/coding-agent/application.py
@@ -0,0 +1,363 @@
+"""Tools the coding agent can call. Each takes typed args and returns a dict."""
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import inspect
+import json
+import os
+from typing import Callable, Optional
+
+from tools import list_files, read_file, run_bash, write_file
+
+from burr.core import State, action, expr, when
+from burr.core.application import ApplicationBuilder
+
+TOOLS = {
+ "list_files": list_files,
+ "read_file": read_file,
+ "write_file": write_file,
+ "run_bash": run_bash,
+}
+
+TYPE_MAP = {str: "string", int: "integer", float: "number", bool: "boolean"}
+
+OPENAI_TOOLS = [
+ {
+ "type": "function",
+ "function": {
+ "name": name,
+ "description": fn.__doc__ or name,
+ "parameters": {
+ "type": "object",
+ "properties": {
+ p.name: {
+ "type": TYPE_MAP.get(p.annotation, "string"),
+ "description": p.name,
+ }
+ for p in inspect.signature(fn).parameters.values()
+ },
+ "required": [
+ p.name
+ for p in inspect.signature(fn).parameters.values()
+ if p.default is inspect.Parameter.empty
+ ],
+ },
+ },
+ }
+ for name, fn in TOOLS.items()
+]
+
+SYSTEM_PROMPT = (
+ "You are a coding agent working inside a project directory. "
+ "Use the tools to inspect and modify files, and to run commands. "
+ "Work one step at a time: look before you edit, and verify changes by running them. "
+ "When the task is complete, reply with a short summary and request no tool."
+)
+
+
+# --- LLM clients -----------------------------------------------------------
+# Both return the same shape:
+# {"content": str | None,
+# "tool_calls": [{"id", "name", "args", "error"?}]}
+# `error` is present (and `args` left empty) when the model emitted arguments
+# that could not be parsed as JSON; the harness turns it into a tool result
+# rather than crashing the run.
+
+
+class OpenAIClient:
+ """Calls OpenAI's chat completions API with tool calling enabled."""
+
+ def __init__(self, model: str = "gpt-4o"):
+ self.model = model
+
+ def __call__(self, messages: list[dict]) -> dict:
+ import openai
+
+ response = openai.chat.completions.create(
+ model=self.model, messages=messages, tools=OPENAI_TOOLS
+ )
+ message = response.choices[0].message
+ calls = []
+ for c in message.tool_calls or []:
+ try:
+ raw = c.function.arguments
+ # A JSON string is what the API contract promises, but some
+ # OpenAI-compatible backends send None when streaming failed to
+ # assemble the arguments -- json.loads(None) raises TypeError,
+ # which is exactly the crash this block exists to prevent.
+ if not isinstance(raw, str):
+ raise TypeError(f"expected a JSON string, got {type(raw).__name__}")
+ args = json.loads(raw)
+ except (json.JSONDecodeError, TypeError) as e:
+ # Models occasionally emit malformed arguments. Carry the error
+ # through as a tool result instead of crashing the run.
+ calls.append(
+ {
+ "id": c.id,
+ "name": c.function.name,
+ "args": {},
+ "error": f"could not parse tool arguments as JSON: {e}",
+ }
+ )
+ else:
+ calls.append({"id": c.id, "name": c.function.name, "args": args})
+ return {"content": message.content, "tool_calls": calls}
+
+
+class ScriptedClient:
+ """Replays a fixed list of responses. Lets the example run without an API key."""
+
+ def __init__(self, responses: list[dict]):
+ self.responses = list(responses)
+ self.index = 0
+
+ def __call__(self, messages: list[dict]) -> dict:
+ if self.index >= len(self.responses):
+ return {"content": "Script exhausted.", "tool_calls": []}
+ response = self.responses[self.index]
+ self.index += 1
+ return response
+
+
+DEFAULT_SCRIPT = [
+ {"content": None, "tool_calls": [{"id": "c1", "name": "list_files", "args": {}}]},
+ {
+ "content": None,
+ "tool_calls": [
+ {
+ "id": "c2",
+ "name": "write_file",
+ "args": {
+ "path": "hello.py",
+ "contents": "print('hello from the agent')\n",
+ },
+ }
+ ],
+ },
+ {
+ "content": None,
+ "tool_calls": [{"id": "c3", "name": "run_bash", "args": {"command": "python hello.py"}}],
+ },
+ {"content": "Created hello.py and confirmed it runs.", "tool_calls": []},
+]
+
+
+def get_client():
+ """Real client when OPENAI_API_KEY is set, otherwise the scripted stand-in."""
+ if os.environ.get("OPENAI_API_KEY"):
+ return OpenAIClient()
+ return ScriptedClient(DEFAULT_SCRIPT)
+
+
+# --- Actions ---------------------------------------------------------------
+
+
+@action(reads=[], writes=["task", "messages", "steps", "done", "final_answer"])
+def human_input(state: State, task: str) -> State:
+ """Takes a task from the user and starts a fresh run."""
+ return state.update(
+ task=task,
+ messages=[
+ {"role": "system", "content": SYSTEM_PROMPT},
+ {"role": "user", "content": task},
+ ],
+ steps=0,
+ done=False,
+ final_answer=None,
+ )
+
+
+@action(reads=["messages"], writes=["messages"])
+def create_prompt(state: State) -> State:
+ """Seam for shaping the prompt before each call -- add file trees, summaries, etc."""
+ return state.update(messages=state["messages"])
+
+
+@action(
+ reads=["messages", "steps"],
+ writes=[
+ "messages",
+ "next_tool",
+ "next_args",
+ "last_tool_call_id",
+ "steps",
+ "done",
+ "final_answer",
+ ],
+)
+def call_llm(state: State, client: Callable) -> State:
+ """Asks the model what to do next: call a tool, or finish."""
+ result = client(state["messages"])
+ calls = result["tool_calls"]
+
+ if not calls:
+ return state.update(
+ messages=state["messages"] + [{"role": "assistant", "content": result["content"]}],
+ next_tool=None,
+ next_args={},
+ last_tool_call_id=None,
+ steps=state["steps"] + 1,
+ done=True,
+ final_answer=result["content"],
+ )
+
+ # One tool per iteration keeps the graph readable; extras are dropped.
+ call = calls[0]
+ assistant_message = {
+ "role": "assistant",
+ "content": result["content"],
+ "tool_calls": [
+ {
+ "id": call["id"],
+ "type": "function",
+ "function": {
+ "name": call["name"],
+ "arguments": json.dumps(call["args"]),
+ },
+ }
+ ],
+ }
+
+ # A tool the model invented, or arguments that failed to parse, would leave
+ # the run with no matching transition. Answer the call with an error and go
+ # back around the loop, so the model gets a chance to correct itself.
+ problem = call.get("error")
+ if problem is None and call["name"] not in TOOLS:
+ problem = f"unknown tool: {call['name']}"
+ if problem is not None:
+ return state.update(
+ messages=state["messages"]
+ + [
+ assistant_message,
+ {
+ "role": "tool",
+ "tool_call_id": call["id"],
+ "content": json.dumps({"error": problem}),
+ },
+ ],
+ next_tool=None,
+ next_args={},
+ last_tool_call_id=None,
+ steps=state["steps"] + 1,
+ done=False,
+ final_answer=None,
+ )
+
+ return state.update(
+ messages=state["messages"] + [assistant_message],
+ next_tool=call["name"],
+ next_args=call["args"],
+ last_tool_call_id=call["id"],
+ steps=state["steps"] + 1,
+ done=False,
+ final_answer=None,
+ )
+
+
+@action(reads=["next_args", "last_tool_call_id", "messages"], writes=["messages"])
+def execute_tool(state: State, tool_function: Callable) -> State:
+ """Runs one tool and feeds its result back to the model.
+
+ A tool that raises is reported back to the model as an error rather than
+ ending the run -- the point of the loop is that the agent sees the failure
+ and tries something else.
+ """
+ try:
+ result = tool_function(**state["next_args"])
+ except Exception as e: # noqa: BLE001 - deliberately broad; the model recovers
+ result = {"error": f"{type(e).__name__}: {e}"}
+ return state.update(
+ messages=state["messages"]
+ + [
+ {
+ "role": "tool",
+ "tool_call_id": state["last_tool_call_id"],
+ "content": json.dumps(result),
+ }
+ ]
+ )
+
+
+@action(reads=["final_answer", "steps", "max_steps"], writes=["final_answer"])
+def respond(state: State) -> State:
+ """Surfaces the answer, or explains that the budget ran out."""
+ if state["final_answer"]:
+ return state.update(final_answer=state["final_answer"])
+ return state.update(
+ final_answer=f"Stopped after {state['steps']} steps without finishing the task."
+ )
+
+
+def application(app_id: Optional[str] = None, max_steps: int = 15, client: Callable = None):
+ """Builds the coding agent application."""
+ client = client or get_client()
+ return (
+ ApplicationBuilder()
+ .with_actions(
+ human_input,
+ create_prompt,
+ respond,
+ call_llm=call_llm.bind(client=client),
+ read_file=execute_tool.bind(tool_function=read_file),
+ write_file=execute_tool.bind(tool_function=write_file),
+ list_files=execute_tool.bind(tool_function=list_files),
+ run_bash=execute_tool.bind(tool_function=run_bash),
+ )
+ .with_transitions(
+ ("human_input", "create_prompt"),
+ ("create_prompt", "call_llm"),
+ ("call_llm", "respond", when(done=True)),
+ ("call_llm", "respond", expr("steps>=max_steps")),
+ # Order matters here in both directions:
+ # * AFTER the done=True branch above: a finished run has no tool
+ # calls, so done=True and next_tool=None hold at the same time,
+ # and Graph.get_next_node returns the first match in insertion
+ # order. Moving this line up would send a finished run back to
+ # create_prompt and spin until steps>=max_steps.
+ # * BEFORE the per-tool transitions below: an unparseable tool
+ # call is answered with an error so the model can retry.
+ ("call_llm", "create_prompt", when(next_tool=None)),
+ ("call_llm", "read_file", when(next_tool="read_file")),
+ ("call_llm", "write_file", when(next_tool="write_file")),
+ ("call_llm", "list_files", when(next_tool="list_files")),
+ ("call_llm", "run_bash", when(next_tool="run_bash")),
+ # Tool results go back through create_prompt, so the prompt-shaping
+ # seam runs before every model call rather than only the first.
+ (["read_file", "write_file", "list_files", "run_bash"], "create_prompt"),
+ ("respond", "human_input"),
+ )
+ .with_state(max_steps=max_steps, steps=0, messages=[], done=False, final_answer=None)
+ .with_identifiers(app_id=app_id)
+ .with_entrypoint("human_input")
+ .with_tracker(project="demo_coding_agent")
+ .build()
+ )
+
+
+if __name__ == "__main__":
+ app = application()
+ # Rendering the graph needs the Graphviz binary, which the example does not
+ # otherwise require. Skip it when unavailable instead of failing the run.
+ try:
+ app.visualize(output_file_path="./statemachine.png")
+ except Exception as e: # noqa: BLE001
+ print(f"Skipping state machine render (Graphviz unavailable: {e})")
+ _, _, state = app.run(
+ halt_after=["respond"],
+ inputs={"task": "Create a hello.py that prints a greeting, then run it."},
+ )
+ print(state["final_answer"])
diff --git a/examples/coding-agent/notebook.ipynb b/examples/coding-agent/notebook.ipynb
new file mode 100644
index 000000000..ce354a34f
--- /dev/null
+++ b/examples/coding-agent/notebook.ipynb
@@ -0,0 +1,716 @@
+{
+ "cells": [
+ {
+ "cell_type": "code",
+ "execution_count": 7,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "# Licensed to the Apache Software Foundation (ASF) under one\n",
+ "# or more contributor license agreements. See the NOTICE file\n",
+ "# distributed with this work for additional information\n",
+ "# regarding copyright ownership. The ASF licenses this file\n",
+ "# to you under the Apache License, Version 2.0 (the\n",
+ "# \"License\"); you may not use this file except in compliance\n",
+ "# with the License. You may obtain a copy of the License at\n",
+ "#\n",
+ "# http://www.apache.org/licenses/LICENSE-2.0\n",
+ "#\n",
+ "# Unless required by applicable law or agreed to in writing,\n",
+ "# software distributed under the License is distributed on an\n",
+ "# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n",
+ "# KIND, either express or implied. See the License for the\n",
+ "# specific language governing permissions and limitations\n",
+ "# under the License."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "# Coding agent\n",
+ "\n",
+ "A minimal coding agent harness: an LLM that reads files, writes files, and runs\n",
+ "shell commands in a loop until the task is done.\n",
+ "\n",
+ "The point is the **loop**. A tool-calling assistant runs one tool and hands back\n",
+ "to the user. An agent keeps going by itself -- read a file, edit it, run the\n",
+ "tests, see them fail, try again -- with no human in between. That is a cycle,\n",
+ "and cycles are what Burr is for.\n",
+ "\n",
+ "This notebook runs without an API key: a scripted client stands in for the model."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "# Imports"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 8,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "import inspect\n",
+ "import json\n",
+ "import os\n",
+ "import subprocess\n",
+ "from typing import Callable, Optional\n",
+ "\n",
+ "from burr.core import State, action, expr, when\n",
+ "from burr.core.application import ApplicationBuilder"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "# The tools\n",
+ "\n",
+ "Four functions. Each takes primitive, type-annotated arguments and returns a\n",
+ "dict -- and never raises. An exception would kill the state machine; an\n",
+ "`{\"error\": ...}` dict goes back to the model, which can then try something else.\n",
+ "\n",
+ "Paths are confined to a workspace directory. Note that `run_bash` can `cd` out\n",
+ "of it: this is a teaching example, not a sandbox."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 9,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "\"\"\"Tools the coding agent can call. Each takes typed args and returns a dict.\"\"\"\n",
+ "# Licensed to the Apache Software Foundation (ASF) under one\n",
+ "# or more contributor license agreements. See the NOTICE file\n",
+ "# distributed with this work for additional information\n",
+ "# regarding copyright ownership. The ASF licenses this file\n",
+ "# to you under the Apache License, Version 2.0 (the\n",
+ "# \"License\"); you may not use this file except in compliance\n",
+ "# with the License. You may obtain a copy of the License at\n",
+ "#\n",
+ "# http://www.apache.org/licenses/LICENSE-2.0\n",
+ "#\n",
+ "# Unless required by applicable law or agreed to in writing,\n",
+ "# software distributed under the License is distributed on an\n",
+ "# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n",
+ "# KIND, either express or implied. See the License for the\n",
+ "# specific language governing permissions and limitations\n",
+ "# under the License.\n",
+ "\n",
+ "import inspect\n",
+ "import json\n",
+ "import os\n",
+ "from typing import Callable, Optional\n",
+ "\n",
+ "import openai\n",
+ "import requests\n",
+ "\n",
+ "from burr.core import State, action, when\n",
+ "from burr.core.application import ApplicationBuilder\n",
+ "\n",
+ "import os\n",
+ "import subprocess\n",
+ "\n",
+ "# Everything is confined to this directory. See README on why this is not a sandbox.\n",
+ "WORKSPACE = os.environ.get(\"CODING_AGENT_WORKSPACE\", \"./workspace\")\n",
+ "\n",
+ "\n",
+ "def _resolve(path: str) -> str:\n",
+ " \"\"\"Resolve a path inside the workspace, refusing anything that escapes it.\"\"\"\n",
+ " root = os.path.abspath(WORKSPACE)\n",
+ " full = os.path.abspath(os.path.join(root, path))\n",
+ " if not full.startswith(root + os.sep) and full != root:\n",
+ " raise ValueError(f\"path escapes workspace: {path}\")\n",
+ " return full\n",
+ "\n",
+ "\n",
+ "def list_files(directory: str = \".\") -> dict:\n",
+ " \"\"\"Lists the files and directories at the given path inside the workspace.\"\"\"\n",
+ " try:\n",
+ " target = _resolve(directory)\n",
+ " return {\"entries\": sorted(os.listdir(target))}\n",
+ " except (ValueError, OSError) as e:\n",
+ " return {\"error\": str(e)}\n",
+ "\n",
+ "\n",
+ "def read_file(path: str) -> dict:\n",
+ " \"\"\"Reads a text file from the workspace and returns its contents.\"\"\"\n",
+ " try:\n",
+ " with open(_resolve(path)) as f:\n",
+ " return {\"path\": path, \"contents\": f.read()}\n",
+ " except (ValueError, OSError) as e:\n",
+ " return {\"error\": str(e)}\n",
+ "\n",
+ "\n",
+ "def write_file(path: str, contents: str) -> dict:\n",
+ " \"\"\"Writes text to a file in the workspace, creating or overwriting it.\"\"\"\n",
+ " try:\n",
+ " full = _resolve(path)\n",
+ " os.makedirs(os.path.dirname(full), exist_ok=True)\n",
+ " with open(full, \"w\") as f:\n",
+ " f.write(contents)\n",
+ " return {\"path\": path, \"bytes_written\": len(contents)}\n",
+ " except (ValueError, OSError) as e:\n",
+ " return {\"error\": str(e)}\n",
+ "\n",
+ "\n",
+ "def run_bash(command: str) -> dict:\n",
+ " \"\"\"Runs a shell command in the workspace and returns its output.\"\"\"\n",
+ " try:\n",
+ " proc = subprocess.run(\n",
+ " command, shell=True, cwd=os.path.abspath(WORKSPACE),\n",
+ " capture_output=True, text=True, timeout=30,\n",
+ " )\n",
+ " return {\n",
+ " \"exit_code\": proc.returncode,\n",
+ " \"stdout\": proc.stdout[-4000:],\n",
+ " \"stderr\": proc.stderr[-4000:],\n",
+ " }\n",
+ " except subprocess.TimeoutExpired:\n",
+ " return {\"error\": \"command timed out after 30s\"}\n",
+ " "
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "# Describing the tools to the model\n",
+ "\n",
+ "The schema is derived from signatures and docstrings, so the tools stay the single source of truth. Only parameters without defaults are marked required."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 10,
+ "id": "dfc10623",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "TOOLS = {\n",
+ " \"list_files\": list_files,\n",
+ " \"read_file\": read_file,\n",
+ " \"write_file\": write_file,\n",
+ " \"run_bash\": run_bash,\n",
+ "}"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 11,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "TYPE_MAP = {str: \"string\", int: \"integer\", float: \"number\", bool: \"boolean\"}\n",
+ "\n",
+ "OPENAI_TOOLS = [\n",
+ " {\n",
+ " \"type\": \"function\",\n",
+ " \"function\": {\n",
+ " \"name\": name,\n",
+ " \"description\": fn.__doc__ or name,\n",
+ " \"parameters\": {\n",
+ " \"type\": \"object\",\n",
+ " \"properties\": {\n",
+ " p.name: {\n",
+ " \"type\": TYPE_MAP.get(p.annotation, \"string\"),\n",
+ " \"description\": p.name,\n",
+ " }\n",
+ " for p in inspect.signature(fn).parameters.values()\n",
+ " },\n",
+ " \"required\": [\n",
+ " p.name\n",
+ " for p in inspect.signature(fn).parameters.values()\n",
+ " if p.default is inspect.Parameter.empty\n",
+ " ],\n",
+ " },\n",
+ " },\n",
+ " }\n",
+ " for name, fn in TOOLS.items()\n",
+ "]"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 12,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "SYSTEM_PROMPT = (\n",
+ " \"You are a coding agent working inside a project directory. \"\n",
+ " \"Use the tools to inspect and modify files, and to run commands. \"\n",
+ " \"Work one step at a time: look before you edit, and verify changes by running them. \"\n",
+ " \"When the task is complete, reply with a short summary and request no tool.\"\n",
+ ")"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "# Two clients\n",
+ "\n",
+ "Both return the same shape, so the rest of the harness doesn't care which is in\n",
+ "use. The scripted client replays a fixed sequence, which is what lets this\n",
+ "notebook -- and CI -- run with no API key."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 13,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "class OpenAIClient:\n",
+ " \"\"\"Calls OpenAI's chat completions API with tool calling enabled.\"\"\"\n",
+ "\n",
+ " def __init__(self, model: str = \"gpt-4o\"):\n",
+ " self.model = model\n",
+ "\n",
+ " def __call__(self, messages: list[dict]) -> dict:\n",
+ " import openai\n",
+ "\n",
+ " response = openai.chat.completions.create(\n",
+ " model=self.model, messages=messages, tools=OPENAI_TOOLS\n",
+ " )\n",
+ " message = response.choices[0].message\n",
+ " calls = [\n",
+ " {\"id\": c.id, \"name\": c.function.name, \"args\": json.loads(c.function.arguments)}\n",
+ " for c in (message.tool_calls or [])\n",
+ " ]\n",
+ " return {\"content\": message.content, \"tool_calls\": calls}\n",
+ "\n",
+ "\n",
+ "class ScriptedClient:\n",
+ " \"\"\"Replays a fixed list of responses. Lets the example run without an API key.\"\"\"\n",
+ "\n",
+ " def __init__(self, responses: list[dict]):\n",
+ " self.responses = list(responses)\n",
+ " self.index = 0\n",
+ "\n",
+ " def __call__(self, messages: list[dict]) -> dict:\n",
+ " if self.index >= len(self.responses):\n",
+ " return {\"content\": \"Script exhausted.\", \"tool_calls\": []}\n",
+ " response = self.responses[self.index]\n",
+ " self.index += 1\n",
+ " return response\n",
+ "\n",
+ "\n",
+ "DEFAULT_SCRIPT = [\n",
+ " {\"content\": None, \"tool_calls\": [{\"id\": \"c1\", \"name\": \"list_files\", \"args\": {}}]},\n",
+ " {\n",
+ " \"content\": None,\n",
+ " \"tool_calls\": [\n",
+ " {\"id\": \"c2\", \"name\": \"write_file\",\n",
+ " \"args\": {\"path\": \"hello.py\", \"contents\": \"print('hello from the agent')\\n\"}}\n",
+ " ],\n",
+ " },\n",
+ " {\"content\": None, \"tool_calls\": [{\"id\": \"c3\", \"name\": \"run_bash\",\n",
+ " \"args\": {\"command\": \"python hello.py\"}}]},\n",
+ " {\"content\": \"Created hello.py and confirmed it runs.\", \"tool_calls\": []},\n",
+ "]\n",
+ "\n",
+ "\n",
+ "def get_client():\n",
+ " \"\"\"Real client when OPENAI_API_KEY is set, otherwise the scripted stand-in.\"\"\"\n",
+ " if os.environ.get(\"OPENAI_API_KEY\"):\n",
+ " return OpenAIClient()\n",
+ " return ScriptedClient(DEFAULT_SCRIPT)"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "# The actions\n",
+ "\n",
+ "`call_llm` is the interesting one. It appends the model's reply to `messages`,\n",
+ "and either sets `done` (no tool requested) or records which tool to run next.\n",
+ "Tool results come back as `role: \"tool\"` messages carrying the `tool_call_id`\n",
+ "they answer -- that is how the model sees what its last action returned."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 14,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "@action(reads=[], writes=[\"task\", \"messages\", \"steps\", \"done\", \"final_answer\"])\n",
+ "def human_input(state: State, task: str) -> State:\n",
+ " \"\"\"Takes a task from the user and starts a fresh run.\"\"\"\n",
+ " return state.update(\n",
+ " task=task,\n",
+ " messages=[\n",
+ " {\"role\": \"system\", \"content\": SYSTEM_PROMPT},\n",
+ " {\"role\": \"user\", \"content\": task},\n",
+ " ],\n",
+ " steps=0,\n",
+ " done=False,\n",
+ " final_answer=None,\n",
+ " )\n",
+ "\n",
+ "\n",
+ "@action(reads=[\"messages\"], writes=[\"messages\"])\n",
+ "def create_prompt(state: State) -> State:\n",
+ " \"\"\"Seam for shaping the prompt before each call -- add file trees, summaries, etc.\"\"\"\n",
+ " return state.update(messages=state[\"messages\"])\n",
+ "\n",
+ "\n",
+ "@action(\n",
+ " reads=[\"messages\", \"steps\"],\n",
+ " writes=[\"messages\", \"next_tool\", \"next_args\", \"last_tool_call_id\",\n",
+ " \"steps\", \"done\", \"final_answer\"],\n",
+ ")\n",
+ "def call_llm(state: State, client: Callable) -> State:\n",
+ " \"\"\"Asks the model what to do next: call a tool, or finish.\"\"\"\n",
+ " result = client(state[\"messages\"])\n",
+ " calls = result[\"tool_calls\"]\n",
+ "\n",
+ " if not calls:\n",
+ " return state.update(\n",
+ " messages=state[\"messages\"] + [{\"role\": \"assistant\", \"content\": result[\"content\"]}],\n",
+ " next_tool=None,\n",
+ " next_args={},\n",
+ " last_tool_call_id=None,\n",
+ " steps=state[\"steps\"] + 1,\n",
+ " done=True,\n",
+ " final_answer=result[\"content\"],\n",
+ " )\n",
+ "\n",
+ " # One tool per iteration keeps the graph readable; extras are dropped.\n",
+ " call = calls[0]\n",
+ " assistant_message = {\n",
+ " \"role\": \"assistant\",\n",
+ " \"content\": result[\"content\"],\n",
+ " \"tool_calls\": [\n",
+ " {\n",
+ " \"id\": call[\"id\"],\n",
+ " \"type\": \"function\",\n",
+ " \"function\": {\"name\": call[\"name\"], \"arguments\": json.dumps(call[\"args\"])},\n",
+ " }\n",
+ " ],\n",
+ " }\n",
+ " return state.update(\n",
+ " messages=state[\"messages\"] + [assistant_message],\n",
+ " next_tool=call[\"name\"],\n",
+ " next_args=call[\"args\"],\n",
+ " last_tool_call_id=call[\"id\"],\n",
+ " steps=state[\"steps\"] + 1,\n",
+ " done=False,\n",
+ " final_answer=None,\n",
+ " )\n",
+ "\n",
+ "\n",
+ "@action(reads=[\"next_args\", \"last_tool_call_id\", \"messages\"], writes=[\"messages\"])\n",
+ "def execute_tool(state: State, tool_function: Callable) -> State:\n",
+ " \"\"\"Runs one tool and feeds its result back to the model.\"\"\"\n",
+ " result = tool_function(**state[\"next_args\"])\n",
+ " return state.update(\n",
+ " messages=state[\"messages\"]\n",
+ " + [\n",
+ " {\n",
+ " \"role\": \"tool\",\n",
+ " \"tool_call_id\": state[\"last_tool_call_id\"],\n",
+ " \"content\": json.dumps(result),\n",
+ " }\n",
+ " ]\n",
+ " )\n",
+ "\n",
+ "\n",
+ "@action(reads=[\"final_answer\", \"steps\", \"max_steps\"], writes=[\"final_answer\"])\n",
+ "def respond(state: State) -> State:\n",
+ " \"\"\"Surfaces the answer, or explains that the budget ran out.\"\"\"\n",
+ " if state[\"final_answer\"]:\n",
+ " return state.update(final_answer=state[\"final_answer\"])\n",
+ " return state.update(\n",
+ " final_answer=f\"Stopped after {state['steps']} steps without finishing the task.\"\n",
+ " )"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "# Wiring the graph\n",
+ "\n",
+ "Two transitions leave `call_llm` for `respond`: the model finished, or the step\n",
+ "budget ran out. Both are listed **before** the tool transitions, because Burr\n",
+ "takes the first matching condition. An agent that picks its own next step can\n",
+ "loop forever, so the budget is not optional.\n",
+ "\n",
+ "Each tool is a separately bound action, so the Burr UI shows exactly which one\n",
+ "ran at each step."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 15,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "def application(app_id: Optional[str] = None, max_steps: int = 15, client: Callable = None):\n",
+ " \"\"\"Builds the coding agent application.\"\"\"\n",
+ " client = client or get_client()\n",
+ " return (\n",
+ " ApplicationBuilder()\n",
+ " .with_actions(\n",
+ " human_input,\n",
+ " create_prompt,\n",
+ " respond,\n",
+ " call_llm=call_llm.bind(client=client),\n",
+ " read_file=execute_tool.bind(tool_function=read_file),\n",
+ " write_file=execute_tool.bind(tool_function=write_file),\n",
+ " list_files=execute_tool.bind(tool_function=list_files),\n",
+ " run_bash=execute_tool.bind(tool_function=run_bash),\n",
+ " )\n",
+ " .with_transitions(\n",
+ " (\"human_input\", \"create_prompt\"),\n",
+ " (\"create_prompt\", \"call_llm\"),\n",
+ " (\"call_llm\", \"respond\", when(done=True)),\n",
+ " (\"call_llm\", \"respond\", expr(\"steps>=max_steps\")),\n",
+ " (\"call_llm\", \"read_file\", when(next_tool=\"read_file\")),\n",
+ " (\"call_llm\", \"write_file\", when(next_tool=\"write_file\")),\n",
+ " (\"call_llm\", \"list_files\", when(next_tool=\"list_files\")),\n",
+ " (\"call_llm\", \"run_bash\", when(next_tool=\"run_bash\")),\n",
+ " ([\"read_file\", \"write_file\", \"list_files\", \"run_bash\"], \"call_llm\"),\n",
+ " (\"respond\", \"human_input\"),\n",
+ " )\n",
+ " .with_state(max_steps=max_steps, steps=0, messages=[], done=False, final_answer=None)\n",
+ " .with_identifiers(app_id=app_id)\n",
+ " .with_entrypoint(\"human_input\")\n",
+ " .with_tracker(project=\"demo_coding_agent\")\n",
+ " .build()\n",
+ " )"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "# Running it"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 16,
+ "metadata": {},
+ "outputs": [
+ {
+ "data": {
+ "image/svg+xml": [
+ "\n",
+ "\n",
+ "\n",
+ "\n",
+ "\n"
+ ],
+ "text/plain": [
+ ""
+ ]
+ },
+ "execution_count": 16,
+ "metadata": {},
+ "output_type": "execute_result"
+ }
+ ],
+ "source": [
+ "app = application()\n",
+ "app.visualize()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 17,
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "Created hello.py and confirmed it runs.\n"
+ ]
+ }
+ ],
+ "source": [
+ "_, _, state = app.run(\n",
+ " halt_after=[\"respond\"],\n",
+ " inputs={\"task\": \"Create a hello.py that prints a greeting, then run it.\"},\n",
+ ")\n",
+ "print(state[\"final_answer\"])"
+ ]
+ }
+ ],
+ "metadata": {
+ "kernelspec": {
+ "display_name": ".venv (3.12.1.final.0)",
+ "language": "python",
+ "name": "python3"
+ },
+ "language_info": {
+ "codemirror_mode": {
+ "name": "ipython",
+ "version": 3
+ },
+ "file_extension": ".py",
+ "mimetype": "text/x-python",
+ "name": "python",
+ "nbconvert_exporter": "python",
+ "pygments_lexer": "ipython3",
+ "version": "3.12.1"
+ }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}
diff --git a/examples/coding-agent/requirements.txt b/examples/coding-agent/requirements.txt
new file mode 100644
index 000000000..e94d27907
--- /dev/null
+++ b/examples/coding-agent/requirements.txt
@@ -0,0 +1,19 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+burr
+openai
diff --git a/examples/coding-agent/statemachine.png b/examples/coding-agent/statemachine.png
new file mode 100644
index 000000000..701ea129c
Binary files /dev/null and b/examples/coding-agent/statemachine.png differ
diff --git a/examples/coding-agent/test_coding_agent.py b/examples/coding-agent/test_coding_agent.py
new file mode 100644
index 000000000..1e2be74bf
--- /dev/null
+++ b/examples/coding-agent/test_coding_agent.py
@@ -0,0 +1,376 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Tests for the coding agent example.
+
+The agent is driven by a `ScriptedClient`, so the whole state machine -- tool
+dispatch, error recovery and the step budget -- is exercised without an API key.
+"""
+
+import json
+import os
+import sys
+
+import pytest
+import tools
+from application import OpenAIClient, ScriptedClient, application
+
+
+@pytest.fixture(autouse=True)
+def workspace(tmp_path, monkeypatch):
+ """Points the tools at an isolated directory for each test.
+
+ ``tools.WORKSPACE`` is read when a tool is called, so patching the module
+ attribute is enough -- tests never touch the real ``./workspace``.
+ """
+ monkeypatch.setattr(tools, "WORKSPACE", str(tmp_path))
+ return tmp_path
+
+
+def run(task: str, script: list[dict], max_steps: int = 15, tmp_path=None):
+ """Runs the agent with a fixed script and returns the final state."""
+ client = ScriptedClient(script)
+ app = application(max_steps=max_steps, client=client)
+ _, _, state = app.run(halt_after=["respond"], inputs={"task": task})
+ return state
+
+
+def tool_result(state: dict, index: int = -1) -> dict:
+ """Returns the JSON payload of the `index`-th tool message in the history."""
+ tool_messages = [m for m in state["messages"] if m["role"] == "tool"]
+ return json.loads(tool_messages[index]["content"])
+
+
+def test_happy_path_runs_to_completion(tmp_path, monkeypatch):
+ """A scripted run calls tools, then stops when the model requests no tool."""
+ monkeypatch.chdir(tmp_path)
+ script = [
+ {
+ "content": None,
+ "tool_calls": [{"id": "c1", "name": "list_files", "args": {}}],
+ },
+ {
+ "content": None,
+ "tool_calls": [
+ {
+ "id": "c2",
+ "name": "write_file",
+ "args": {"path": "hello.py", "contents": "print('hi')\n"},
+ }
+ ],
+ },
+ {"content": "Done.", "tool_calls": []},
+ ]
+ state = run("write hello.py", script)
+
+ assert state["done"] is True
+ assert state["final_answer"] == "Done."
+ # three model calls, each counted once
+ assert state["steps"] == 3
+ assert (tmp_path / "hello.py").read_text() == "print('hi')\n"
+ # every tool call got exactly one result message
+ assert len([m for m in state["messages"] if m["role"] == "tool"]) == 2
+
+
+def test_step_budget_terminates_the_run(tmp_path, monkeypatch):
+ """The agent stops once `max_steps` is reached instead of looping forever."""
+ monkeypatch.chdir(tmp_path)
+ # a script that would never terminate on its own
+ script = [
+ {
+ "content": None,
+ "tool_calls": [{"id": f"c{i}", "name": "list_files", "args": {}}],
+ }
+ for i in range(10)
+ ]
+ state = run("list forever", script, max_steps=3)
+
+ assert state["steps"] == 3
+ assert "without finishing" in state["final_answer"]
+ assert "3 steps" in state["final_answer"]
+
+
+def test_unknown_tool_is_reported_back_to_the_model(tmp_path, monkeypatch):
+ """An invented tool name becomes an error result and the loop continues."""
+ monkeypatch.chdir(tmp_path)
+ script = [
+ {
+ "content": None,
+ "tool_calls": [{"id": "c1", "name": "delete_everything", "args": {}}],
+ },
+ {"content": "Recovered.", "tool_calls": []},
+ ]
+ state = run("do something", script)
+
+ assert state["done"] is True
+ assert state["final_answer"] == "Recovered."
+ assert "unknown tool" in tool_result(state)["error"]
+ # the failed call did not abort the run
+ assert state["steps"] == 2
+
+
+def test_malformed_tool_arguments_do_not_crash_the_run(tmp_path, monkeypatch):
+ """Arguments that failed to parse surface as an error the model can react to."""
+ monkeypatch.chdir(tmp_path)
+ script = [
+ {
+ "content": None,
+ "tool_calls": [
+ {
+ "id": "c1",
+ "name": "write_file",
+ "args": {},
+ "error": "could not parse tool arguments as JSON",
+ }
+ ],
+ },
+ {"content": "Fixed.", "tool_calls": []},
+ ]
+ state = run("write a file", script)
+
+ assert state["done"] is True
+ assert "could not parse" in tool_result(state)["error"]
+
+
+def test_tool_failure_is_reported_back_to_the_model(tmp_path, monkeypatch):
+ """A tool that raises returns an error result rather than killing the run."""
+ monkeypatch.chdir(tmp_path)
+ script = [
+ {
+ "content": None,
+ "tool_calls": [
+ {
+ "id": "c1",
+ "name": "read_file",
+ "args": {"path": "does-not-exist.txt"},
+ }
+ ],
+ },
+ {"content": "That file is missing.", "tool_calls": []},
+ ]
+ state = run("read a file", script)
+
+ assert state["done"] is True
+ assert state["final_answer"] == "That file is missing."
+ assert "error" in tool_result(state)
+
+
+@pytest.mark.parametrize(
+ "path",
+ ["../outside.txt", "sub/../../outside.txt", "/etc/passwd"],
+)
+def test_reads_are_confined_to_the_workspace(tmp_path, monkeypatch, path):
+ """Paths that escape the project directory are refused."""
+ monkeypatch.chdir(tmp_path)
+ script = [
+ {
+ "content": None,
+ "tool_calls": [{"id": "c1", "name": "read_file", "args": {"path": path}}],
+ },
+ {"content": "Stopped.", "tool_calls": []},
+ ]
+ state = run("read outside", script)
+
+ result = tool_result(state)
+ assert "error" in result
+ assert "outside" in result["error"].lower() or "workspace" in result["error"].lower()
+
+
+def test_writes_are_confined_to_the_workspace(tmp_path, monkeypatch):
+ """A write aimed outside the project directory is refused, and lands nowhere."""
+ monkeypatch.chdir(tmp_path)
+ script = [
+ {
+ "content": None,
+ "tool_calls": [
+ {
+ "id": "c1",
+ "name": "write_file",
+ "args": {"path": "../escaped.txt", "contents": "nope"},
+ }
+ ],
+ },
+ {"content": "Stopped.", "tool_calls": []},
+ ]
+ state = run("write outside", script)
+
+ assert "error" in tool_result(state)
+ assert not (tmp_path.parent / "escaped.txt").exists()
+
+
+def test_tool_results_are_fed_back_to_the_model(tmp_path, monkeypatch):
+ """The scripted client sees each tool result before deciding what to do next."""
+ monkeypatch.chdir(tmp_path)
+ (tmp_path / "data.txt").write_text("hello")
+ seen = []
+
+ class RecordingClient:
+ def __init__(self):
+ self.index = 0
+
+ def __call__(self, messages):
+ seen.append([m["role"] for m in messages])
+ self.index += 1
+ if self.index == 1:
+ return {
+ "content": None,
+ "tool_calls": [{"id": "c1", "name": "read_file", "args": {"path": "data.txt"}}],
+ }
+ return {"content": "Read it.", "tool_calls": []}
+
+ app = application(client=RecordingClient())
+ _, _, state = app.run(halt_after=["respond"], inputs={"task": "read data.txt"})
+
+ assert state["final_answer"] == "Read it."
+ # the second model call must include the tool result from the first
+ assert "tool" in seen[1]
+ assert tool_result(state)["contents"] == "hello"
+
+
+# --- OpenAIClient argument parsing -----------------------------------------
+# The scripted-client tests above hand `application` an already-parsed call,
+# which means the parsing code itself was never exercised. These tests drive
+# `OpenAIClient` with a stand-in `openai` module so the real path is covered.
+
+
+class _FakeFunction:
+ def __init__(self, name, arguments):
+ self.name = name
+ self.arguments = arguments
+
+
+class _FakeToolCall:
+ def __init__(self, call_id, name, arguments):
+ self.id = call_id
+ self.function = _FakeFunction(name, arguments)
+
+
+def _fake_openai(tool_calls):
+ """Builds a minimal stand-in for the `openai` module."""
+
+ class _Completions:
+ @staticmethod
+ def create(**_kwargs):
+ message = type("Message", (), {"content": None, "tool_calls": tool_calls})()
+ choice = type("Choice", (), {"message": message})()
+ return type("Response", (), {"choices": [choice]})()
+
+ chat = type("Chat", (), {"completions": _Completions()})()
+ return type("Module", (), {"chat": chat})()
+
+
+@pytest.mark.parametrize(
+ "arguments, expected",
+ [
+ # json.loads(None) raises TypeError -- the crash this code exists to stop
+ (None, "expected a JSON string, got NoneType"),
+ ("{not json}", "could not parse tool arguments as JSON"),
+ ('{"path": "a.txt"}', None),
+ ],
+)
+def test_openai_client_survives_unparseable_arguments(monkeypatch, arguments, expected):
+ """Bad tool arguments become an error result instead of killing the run."""
+ monkeypatch.setitem(
+ sys.modules,
+ "openai",
+ _fake_openai([_FakeToolCall("c1", "write_file", arguments)]),
+ )
+
+ call = OpenAIClient()([{"role": "user", "content": "hi"}])["tool_calls"][0]
+
+ assert call["id"] == "c1" and call["name"] == "write_file"
+ if expected is None:
+ assert "error" not in call
+ assert call["args"] == {"path": "a.txt"}
+ else:
+ assert expected in call["error"]
+ assert call["args"] == {}
+
+
+# --- workspace confinement edges -------------------------------------------
+
+
+@pytest.mark.parametrize(
+ ("root", "full", "expected"),
+ [
+ # the root itself counts as inside
+ ("/ws", "/ws", True),
+ # a plain descendant
+ ("/ws", "/ws/sub", True),
+ # a sibling sharing a name prefix must NOT pass — this is what a naive
+ # `startswith(root + os.sep)` test got right but `-sibling` + `..`
+ # tricks could still confuse
+ ("/ws", "/ws-sibling", False),
+ ("/ws", "/ws-sibling/sub", False),
+ # escaping upwards
+ ("/ws", "/", False),
+ ("/ws/sub", "/ws", False),
+ # unrelated trees have no common ancestor
+ ("C:\\ws", "D:\\ws\\sub", False),
+ ],
+)
+def test_is_within_compares_path_components(root, full, expected):
+ """Containment is decided per path component, not per character.
+
+ Tested through `_is_within` rather than `_resolve` so the assertions do not
+ depend on the case behaviour of the filesystem this suite runs on — the
+ old test went through `_resolve` and was therefore a false positive on
+ macOS, where `realpath` already normalises case and `normcase` never ran.
+ """
+ assert tools._is_within(root, full) is expected
+
+
+@pytest.mark.skipif(os.name != "nt", reason="drive-letter comparison is Windows-only")
+def test_is_within_is_case_insensitive_on_windows():
+ """On Windows the containment check must ignore case.
+
+ `ntpath.commonpath` normcases before comparing, which is the only reason
+ the case difference below is tolerated.
+ """
+ assert tools._is_within("C:\\WS", "c:\\ws\\sub") is True
+
+
+def test_resolve_accepts_the_workspace_root(tmp_path, monkeypatch):
+ """The workspace itself is inside the workspace."""
+ monkeypatch.setattr(tools, "WORKSPACE", str(tmp_path))
+ assert tools._resolve(".") == os.path.realpath(str(tmp_path))
+
+
+def test_resolve_rejects_a_sibling_with_a_shared_prefix(tmp_path, monkeypatch):
+ """`-sibling` must not pass a naive prefix test against ``."""
+ monkeypatch.setattr(tools, "WORKSPACE", str(tmp_path))
+ sibling = tmp_path.parent / (tmp_path.name + "-sibling")
+ sibling.mkdir()
+
+ with pytest.raises(ValueError, match="escapes workspace"):
+ tools._resolve(os.path.relpath(sibling, tmp_path))
+
+
+def test_resolve_tolerates_case_differences(tmp_path, monkeypatch):
+ """`realpath` may report a different case than WORKSPACE.
+
+ This exercises the real filesystem, so it is skipped where the filesystem is
+ case-sensitive. Note what it does NOT prove: on macOS `realpath` already
+ returns the on-disk case, so this passes there without any normalisation
+ (see `test_is_within_*` for the containment rule itself).
+ """
+ swapped = str(tmp_path).swapcase()
+ if swapped == str(tmp_path) or not os.path.isdir(swapped):
+ pytest.skip("needs a case-insensitive filesystem")
+
+ monkeypatch.setattr(tools, "WORKSPACE", swapped)
+ assert tools._resolve("sub") == os.path.join(os.path.realpath(str(tmp_path)), "sub")
diff --git a/examples/coding-agent/tools.py b/examples/coding-agent/tools.py
new file mode 100644
index 000000000..f819adeb4
--- /dev/null
+++ b/examples/coding-agent/tools.py
@@ -0,0 +1,126 @@
+"""Tools the coding agent can call. Each takes typed args and returns a dict."""
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import os
+import subprocess
+
+# Everything is confined to this directory. See README on why this is not a sandbox.
+WORKSPACE = os.environ.get("CODING_AGENT_WORKSPACE", "./workspace")
+
+
+def _is_within(root: str, full: str) -> bool:
+ """True when ``full`` is ``root`` itself or a descendant of it.
+
+ Split out from :func:`_resolve` so the containment rule can be tested
+ directly, without depending on the case behaviour of the filesystem the
+ test happens to run on.
+
+ ``os.path.commonpath`` is used instead of a string prefix test because it
+ compares path *components*: ``C:\\ws`` is not an ancestor of
+ ``C:\\ws-sibling`` even though the latter starts with the former. It is
+ case-insensitive on Windows (``ntpath`` normcases before comparing) and
+ case-sensitive on POSIX, which is what each platform's filesystem does.
+ """
+ if full == root:
+ return True
+ try:
+ common = os.path.commonpath([root, full])
+ except ValueError:
+ # Different drives (Windows) or a mix of absolute and relative paths:
+ # there is no common ancestor, so `full` cannot be inside `root`.
+ return False
+ # `commonpath` returns a normalised path (and, on Windows, a normcased
+ # one), so normalise the root the same way before comparing — otherwise
+ # `/ws` would not match the `\ws` that `ntpath` hands back.
+ return os.path.normcase(common) == os.path.normcase(os.path.normpath(root))
+
+
+def _resolve(path: str) -> str:
+ """Resolve a path inside the workspace, refusing anything that escapes it.
+
+ Resolves symlinks (``realpath`` rather than ``abspath``) so that a link
+ placed inside the workspace cannot be followed out of it.
+
+ Containment is decided by :func:`_is_within`, which compares path
+ components and is case-insensitive exactly where the platform is. Windows
+ needs that (``realpath`` may return a different case than ``WORKSPACE``);
+ on macOS ``realpath`` already reports the on-disk case, so both sides agree
+ without any extra normalisation — ``os.path.normcase`` is a no-op on POSIX
+ and could not have provided it anyway.
+ """
+ root = os.path.realpath(WORKSPACE)
+ full = os.path.realpath(os.path.join(root, path))
+ if not _is_within(root, full):
+ raise ValueError(f"path escapes workspace: {path}")
+ return full
+
+
+def list_files(directory: str = ".") -> dict:
+ """Lists the files and directories at the given path inside the workspace."""
+ try:
+ target = _resolve(directory)
+ return {"entries": sorted(os.listdir(target))}
+ except (ValueError, OSError) as e:
+ return {"error": str(e)}
+
+
+def read_file(path: str) -> dict:
+ """Reads a text file from the workspace and returns its contents."""
+ try:
+ with open(_resolve(path)) as f:
+ return {"path": path, "contents": f.read()}
+ except (ValueError, OSError) as e:
+ return {"error": str(e)}
+
+
+def write_file(path: str, contents: str) -> dict:
+ """Writes text to a file in the workspace, creating or overwriting it."""
+ try:
+ full = _resolve(path)
+ os.makedirs(os.path.dirname(full), exist_ok=True)
+ with open(full, "w") as f:
+ f.write(contents)
+ return {"path": path, "bytes_written": len(contents)}
+ except (ValueError, OSError) as e:
+ return {"error": str(e)}
+
+
+def run_bash(command: str) -> dict:
+ """Runs a shell command in the workspace and returns its output."""
+ try:
+ workspace = os.path.realpath(WORKSPACE)
+ # The workspace may not exist yet -- a model that starts by running a
+ # command rather than writing a file should still get a useful answer.
+ os.makedirs(workspace, exist_ok=True)
+ proc = subprocess.run(
+ command,
+ shell=True,
+ cwd=workspace,
+ capture_output=True,
+ text=True,
+ timeout=30,
+ )
+ return {
+ "exit_code": proc.returncode,
+ "stdout": proc.stdout[-4000:],
+ "stderr": proc.stderr[-4000:],
+ }
+ except subprocess.TimeoutExpired:
+ return {"error": "command timed out after 30s"}
+ except OSError as e:
+ return {"error": str(e)}
diff --git a/pyproject.toml b/pyproject.toml
index 7fba971cc..7cda7dccb 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -308,6 +308,7 @@ exclude = [
"examples/validate_examples.py",
"examples/fastapi_mount_example.py",
"examples/adaptive-crag/**",
+ "examples/coding-agent/**",
"examples/conversational-rag/**",
"examples/custom-serde/**",
"examples/deployment/**",