Trace a nested agent tool call through every outcome

Run and inspect the same nested tool workflow in three states: success, a structured lookup failure and cancellation. A fixed stock table keeps the result reproducible. Then add a model while preserving the visible action input, tool result and recommendation branch.

Make a real call in Studio#

Select Open the live Studio demo. It opens the public demo endpoint's Actions tab without requiring an installation or account. In the action list, choose split_lines, enter these two lines in the text field and select Call:

first line
second line

The lines output contains two values. Inspect the call's status and duration, then open Wire to see the messages that carried it. This is a new remote execution, unlike the landing's recorded video. For a longer workload, choose text_to_image and supply the request from the image walkthrough.

Start with the questions a debugger should answer:

• What inputs did this action receive?
• Which output values arrived, and when?
• Did it finish, fail or get cancelled?
• Which logs explain the application decision?

Studio uses the demo service's actual actions. It is useful for seeing Studio; the local workflow below is useful for understanding and changing the code.

Reproduce an order recommendation#

Start with a known stock table so the same input produces the same result on every run. You can check the tool and recommendation logic before adding a model that chooses the tool’s arguments.

Use Run this guide above in an active Python 3.11+ virtual environment. It downloads the complete workflow and runs its successful notebook case.

No model or API key is involved in this local fixture. A registered check-order action calls a nested lookup_stock tool, emits progress and returns a recommendation:

Checking stock for notebook…
Preparing recommendation 1/3
Preparing recommendation 2/3
Preparing recommendation 3/3
Ready to order: notebook has 12 in stock.
[completed]

The deterministic workflow isolates the runtime behavior an agent's tools need. Verify the input, tool call and result before introducing model-selected arguments.

Trace the pen recommendation#

The stock table contains twelve notebooks and zero pens. Follow the pen input from the lookup to the recommendation branch: zero stock should lead to an alternative, while a positive quantity should lead to “Ready to order.”

The first registered function is an ordinary lookup:

@REGISTRY.action(output="quantity")
async def lookup_stock(item: str) -> int:
    await asyncio.sleep(0.6)
    stock = {"notebook": 12, "pen": 0}
    if item not in stock:
        raise Status(
            code=StatusCode.NOT_FOUND,
            message=f"No stock record for {item!r}.",
            details=[{"item": item, "known_items": sorted(stock)}],
        ).to_exception()
    return stock[item]

The delay makes an in-flight call observable. notebook is available; pen is sold out; an unknown key raises a structured NOT_FOUND status with the requested item in its details.

The parent creates a nested action with a runtime parent-child relationship:

tool = action.make_nested("lookup_stock", propagate_io=False)
await tool["item"].finalize(item)
tool.run()
try:
    quantity = await tool["quantity"].consume(int)
    await tool.wait()
except StatusException:
    await tool.wait()
    raise
finally:
    tool.cancel()

The parent consumes the quantity output and waits for the child status. propagate_io=False keeps the tool's raw ports local; the parent chooses the public information to emit. Two explicit application logs record the tool request and returned stock. The request is logged before the nested call; the returned quantity is logged after it succeeds:

await action.log(f"lookup_stock(item={item!r})", channel="tool")

After await tool.wait():

await action.log(
    {"item": item, "quantity": quantity}, channel="tool-result"
)

The example writes these decision logs explicitly. A11 records action invocations and explicit logs; arbitrary Python activity remains ordinary application code.

After three progress messages, one ordinary branch chooses the recommendation:

decision = "Ready to order" if quantity else "Choose an alternative"
await action["updates"].finalize(f"{decision}: {item} has {quantity} in stock.")

Run python debug_workflow.py pen. You should get Choose an alternative: pen has 0 in stock. That is a successful action with a negative business outcome. “Out of stock” is not a runtime exception.

Handle a missing item differently from a cancelled lookup#

An unknown item needs a correction to the request or stock data. A lookup cancelled by the user needs to stop without reporting a misleading inventory problem. Exercise both outcomes so the caller can distinguish them.

Ask for a missing record:

python debug_workflow.py missing

Expected final message:

[NOT_FOUND] No stock record for 'missing'.

The terminal caller reads updates, but it also checks the final status when a stream raises:

try:
    async for update in action["updates"]:
        print(update, flush=True)
    await action.wait()
    print("[completed]")
except StatusException:
    await action.wait()
    raise

That second wait() preserves the underlying tool error if the output stream failed first. The tool fails before the parent starts preparing its recommendation. The CLI formats that final status as [NOT_FOUND] with the missing item’s name.

Now cancel during the stock lookup:

python debug_workflow.py --cancel-after 0.2

Expected final message:

[CANCELLED] Action was cancelled

The timer uses the same cancellation call a Stop button would use:

async def stop_later():
    await asyncio.sleep(args.cancel_after)
    action.cancel()

timer = (
    asyncio.create_task(stop_later())
    if args.cancel_after is not None
    else None
)

The caller’s finally block cancels the action and retires that timer so it cannot outlive the invocation:

finally:
    action.cancel()
    if timer is not None:
        timer.cancel()
        await asyncio.gather(timer, return_exceptions=True)

Both deliberate cases exit unsuccessfully and omit [completed]. Run the default command again to verify a fresh invocation succeeds. Cancellation does not undo an external side effect; this fixture only reads an in-memory dictionary, so it is a useful place to learn the lifecycle.

Inspect the tool input and result in Studio#

When a recommendation is wrong, terminal progress alone may not show the tool’s arguments or return value. Serve the same workflow to Studio so you can inspect those records alongside the result.

Keep the file unchanged and run:

a11 serve debug_workflow.py --ws --ws-host 127.0.0.1 --ws-port 18012 --ws-path /ws

Leave that terminal running. For local development, open local Studio at http://127.0.0.1:3000/studio, sign in if prompted, and find Direct connection, enter the URL and select Connect with ws://127.0.0.1:18012/ws. Local Studio must already be running; the Python serve command starts the action server, not the website. The public Studio demo above is the no-setup option. An HTTPS page may block an insecure local WebSocket.

In the action explorer, choose check-order, enter notebook for item and run it. Compare the inputs, streamed updates, tool logs, timing and final status to the terminal result. Then use missing, and finally cancel a fresh run while it is still producing progress. The action schema creates the input and output interface; this example does not build a separate form for Studio.

If you choose lookup_stock directly, you can test the tool without its parent. That makes the debugging question precise: is the problem in the tool itself, or in the policy that consumes its result?

Locate the error before changing the prompt#

For an unexpected “Choose an alternative,” check the records in order:

• If the tool input is pen but the user asked for a notebook, investigate how the caller chose the argument.
• If the input is notebook but the lookup returns zero, investigate the lookup and its data.
• If the lookup returns twelve but the recommendation still says “Choose an alternative,” inspect the branch consuming that result.

Each case points to different code. Changing a model prompt would not repair an incorrect stock table or an inverted branch.

Studio exposes action inputs, outputs, logs, timing, cancellation and wire traffic. The workload uses a native parent-child relationship. Execution-tree and replay views are outside the current walkthrough; application logs provide the decision context beside the protocol traffic.

The telemetry guide describes runtime tracing and root-context setup. Model and tool interactions become inspectable when the application represents them as actions or logs. Add a model after this fixture works, then compare the model's requested arguments to the tool input instead of guessing from the final answer.

Complete source#

Download debug_workflow.py. It contains both registered actions, the caller and the cancellation timer.

"""Run a reproducible nested tool workflow through success, failure, or cancellation."""

import argparse
import asyncio

import a11
from a11.status import Status, StatusCode, StatusException

REGISTRY = a11.ActionRegistry()


@REGISTRY.action(output="quantity")
async def lookup_stock(item: str) -> int:
    await asyncio.sleep(0.6)
    stock = {"notebook": 12, "pen": 0}
    if item not in stock:
        raise Status(
            code=StatusCode.NOT_FOUND,
            message=f"No stock record for {item!r}.",
            details=[{"item": item, "known_items": sorted(stock)}],
        ).to_exception()
    return stock[item]


SCHEMA = a11.ActionSchema(
    name="check-order",
    inputs={
        "item": a11.ActionPortSchema(
            "item", "text/plain", typeinfo=str, unary=True, required=True
        )
    },
    outputs={
        "updates": a11.ActionPortSchema(
            "updates", "text/plain", typeinfo=str, required=True
        )
    },
)


async def check_order(action):
    item = await action["item"].consume(str)
    await action["updates"].put(f"Checking stock for {item}…")
    await action.log(f"lookup_stock(item={item!r})", channel="tool")
    tool = action.make_nested("lookup_stock", propagate_io=False)
    await tool["item"].finalize(item)
    tool.run()
    try:
        quantity = await tool["quantity"].consume(int)
        await tool.wait()
    except StatusException:
        await tool.wait()
        raise
    finally:
        tool.cancel()
    await action.log(
        {"item": item, "quantity": quantity}, channel="tool-result"
    )
    for step in range(3):
        await action["updates"].put(f"Preparing recommendation {step + 1}/3")
        await asyncio.sleep(0.5)
    decision = "Ready to order" if quantity else "Choose an alternative"
    await action["updates"].finalize(
        f"{decision}: {item} has {quantity} in stock."
    )


REGISTRY.register("check-order", SCHEMA, check_order)


async def main(args):
    action = REGISTRY.make_action("check-order").run()
    await action["item"].finalize(args.item)

    async def stop_later():
        await asyncio.sleep(args.cancel_after)
        action.cancel()

    timer = (
        asyncio.create_task(stop_later())
        if args.cancel_after is not None
        else None
    )
    try:
        async for update in action["updates"]:
            print(update, flush=True)
        await action.wait()
        print("[completed]")
    except StatusException:
        await action.wait()
        raise
    finally:
        action.cancel()
        if timer is not None:
            timer.cancel()
            await asyncio.gather(timer, return_exceptions=True)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("item", nargs="?", default="notebook")
    parser.add_argument("--cancel-after", type=float)
    args = parser.parse_args()
    try:
        asyncio.run(main(args))
    except KeyboardInterrupt:
        print("\nStopped.")

    except StatusException as exc:
        raise SystemExit(f"[{exc.status.code.name}] {exc.status.message}")

The public demo's service is implemented in web_demos_server.py; its model-backed research composition is explained in the research walkthrough.