Notebooks · Workspace, live UI and observability · 76

Apps, analysis, and the whole UI — end to end

Name: SHIPIT Agent
Author: SHIPIT

The reference product's screenshot is a chat on the left and a thing on the right: a revenue dashboard, a spreadsheet, a workflow. The chat is where you ask; the thing is what you keep.

Open in Colab View on GitHub 23 code cells · 22 with saved output

Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.

1 · Setup

Same guard as the other notebooks: this repo on the path, and an assertion if some other shipit_agent wins.

In [ ]

python

import sys
from pathlib import Path


def repo_root(start: Path) -> Path:
    for candidate in [start, *start.parents]:
        if (candidate / "shipit_agent" / "__init__.py").exists():
            return candidate
    raise RuntimeError(f"Could not locate the shipit_agent repo from {start}")


REPO = repo_root(Path.cwd())
sys.path.insert(0, str(REPO))
for name in [n for n in sys.modules if n.startswith("shipit_agent")]:
    del sys.modules[name]

import shipit_agent

loaded = Path(shipit_agent.__file__).resolve().parent.parent
assert loaded == REPO.resolve(), f"wrong shipit_agent loaded: {loaded}"
print("repo    :", REPO)
print("version :", shipit_agent.__version__)

Out

repo    : ~/Documents/MYWORK/ai_developer/others/shipit_agent
version : 1.1.0

In [2]

python

import importlib.util

PROVIDER = Path(
    "~/Documents/MYWORK/AFTDRK/CACHE/DRK_CACHE_BACK"
    "/drk_cache/llm/bedrock_mantle_provider.py"
)
MODEL = "bedrock-mantle/google.gemma-4-26b-a4b"

spec = importlib.util.spec_from_file_location("bedrock_mantle_provider", PROVIDER)
module = importlib.util.module_from_spec(spec)
sys.modules["bedrock_mantle_provider"] = module
spec.loader.exec_module(module)
module.ensure_registered()

from shipit_agent.llms import LiteLLMChatLLM


def llm():
    return LiteLLMChatLLM(model=MODEL)


import boto3
print("identity:", boto3.client("sts").get_caller_identity()["Arn"])
print("model   :", MODEL)

Out

identity: arn:aws:iam::275210565507:user/0x99-bedrock
model   : bedrock-mantle/google.gemma-4-26b-a4b

The data

A quarter of bookings, as a CSV — the same shape as the "Revenue by region" panel in the reference UI, small enough to check by eye.

In [3]

python

import tempfile

WORKSPACE = Path(tempfile.mkdtemp(prefix="revenue_"))
(WORKSPACE / "bookings.csv").write_text(
    "region,rep,amount,month\n"
    "EMEA,Dana Kim,120000,Apr\n"
    "EMEA,Luis Marin,84000,May\n"
    "AMER,Sam Osei,240000,Apr\n"
    "AMER,Priya Raman,156000,May\n"
    "AMER,Dana Kim,98000,Jun\n"
    "APAC,Maximo Guk,64000,May\n"
    "APAC,Kenton Varda,45000,Jun\n"
)
(WORKSPACE / "targets.csv").write_text(
    "region,target\nEMEA,200000\nAMER,450000\nAPAC,150000\n"
)
print("workspace:", WORKSPACE)
print((WORKSPACE / "bookings.csv").read_text())

Out

workspace: /var/folders/bd/pq4lv0q52pv59m8pthkn60sc0000gn/T/revenue_eukt2id_
region,rep,amount,month
EMEA,Dana Kim,120000,Apr
EMEA,Luis Marin,84000,May
AMER,Sam Osei,240000,Apr
AMER,Priya Raman,156000,May
AMER,Dana Kim,98000,Jun
APAC,Maximo Guk,64000,May
APAC,Kenton Varda,45000,Jun

In [4]

python

from shipit_agent import Agent
from shipit_agent.builtins import get_builtin_tool_map
from shipit_agent.permissions import PermissionDecision, PermissionEngine

TOOLS = get_builtin_tool_map(llm=None, project_root=str(WORKSPACE))
READERS = [TOOLS[n] for n in ("read_file", "glob_files", "grep_files")]
APPS = [TOOLS[n] for n in
        ("list_blueprints", "create_app", "set_app_binding", "use_app")]

ALLOWED = [t.name for t in (*READERS, *APPS)]
POLICY = PermissionEngine(allow=ALLOWED, default_decision=PermissionDecision.DENY)

print("tools :", ALLOWED)

Out

tools : ['read_file', 'glob_files', 'grep_files', 'list_blueprints', 'create_app', 'set_app_binding', 'use_app']

2 · The four app tools

create_app writes a program into the workspace. use_app runs it. In between, set_app_binding decides what it is allowed to reach — an app sees only what you wire, never the agent's whole environment.

They ship as builtins, storing under <project>/.shipit/apps, so an app outlives the run that wrote it.

In [5]

python

from shipit_agent.tools.base import ToolContext

ctx = ToolContext(prompt="", state={})
print(TOOLS["list_blueprints"].run(ctx).text)

Out

Blueprints:
- `csv_summary` — CSV summary: Reads a CSV and counts rows, optionally grouped by a column.
- `dashboard` — Dashboard: Headline numbers and a bar chart from records — the thing you send someone who asked where revenue landed.
- `page` — HTML page: Renders a self-contained HTML page from records and writes it to disk — something you can send someone.
- `report` — Markdown report: Turns a list of records into a Markdown table.
- `sheet` — Spreadsheet: Records as a spreadsheet: column letters, row numbers, a frozen header, and cells you can flag.
- `workflow` — Workflow diagram: A pipeline as boxes and connectors, with the agent step marked and a live log underneath.

3 · The agent builds an analysis app

Not a script it throws away — an app, with a name, saved where you can run it again. The prompt says what it should do, not how to write it.

In [6]

python

from shipit_agent.narrate import render_tree

analyst = Agent(
    llm=llm(), tools=[*READERS, *APPS], permissions=POLICY,
    auto_use_skills=False, max_iterations=10,
)

BUILD = (
    "Create an app named revenue_by_region. Write its app.py yourself: "
    "run(input, env) must read the CSV at input['path'], total the `amount` "
    "column per `region`, and return {'totals': {...}, 'grand_total': N}. "
    "Then use_app it on path 'bookings.csv' and tell me the totals."
)

build_events = list(analyst.stream(BUILD))
print(render_tree(build_events, model="gemma-4-26b", detail=True, output_lines=10))

Out

Agent started
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed
│     ↳ "name"='revenue_by_region', "title"='Revenue by Region App', "description"='Calculates total revenue per region from a CSV file.', "code="<code>import csv<ol><l
│       Invalid app name '': use lower snake case, 2-41 characters, starting with a letter.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed
│     ↳ "name"='revenue_by_region', "title"='Revenue by Region App', "description"='Calculates total revenue per region from a CSV file.', "code="<code>import csv\\n\\n
│       Invalid app name '': use lower snake case, 2-41 characters, starting with a letter.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  0ms
│     ↳ "name"='revenue_by_region', "title"='Revenue by Region App', "code="many users often fail to provide the python code in a way that is correctly interpreted if t
│       Invalid app name '': use lower snake case, 2-41 characters, starting with a letter.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  0ms
│     ↳ "name"='revenue_by_region', "title"='Revenue by Region App', "code="bold(import csv\\n\\ndef run(input, env):\\n    path = input.get('path')\\n    if not path:\
│       Invalid app name '': use lower snake case, 2-41 characters, starting with a letter.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  0ms
│     ↳ "name"='revenue_by_region', "title"='Revenue by Region App', "code="``import csv\\n\\ndef run(input, env):\\n    path = input.get('path')\\n    if not path:\\n 
│       Invalid app name '': use lower snake case, 2-41 characters, starting with a letter.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  1ms
│     ↳ name='revenue_by_region', title='Revenue by Region App', description='Calculates total revenue per region from a CSV file.', code:="import csv\n\ndef run(input,
│       An app needs app.py defining `run(input, env)` — pass it in files, or start from a blueprint.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  3ms
│     ↳ name='revenue_by_region', title='Revenue by Region App', code:="import csv\n\ndef run(input, env):\n    path = input.get('path')\n    if not path:\n        retu
│       An app needs app.py defining `run(input, env)` — pass it in files, or start from a blueprint.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  4ms
│     ↳ name='revenue_by_region', title='Revenue by Region App', code="import csv\n\ndef run(input, env):\n    path = input.get('path')\n    if not path:\n        retur
… 27 more lines

What it wrote

The app is a file on disk. Read it — this is the thing you keep.

In [7]

python

from shipit_agent.apps import AppStore, run_app

store = AppStore(WORKSPACE / ".shipit" / "apps", workdir=WORKSPACE)
for app in store.list():
    print(app.describe())
    print()
    print(app.entrypoint.read_text())

Out

Revenue by Region App (`revenue_by_region`)
files    : app.json, app.py
bindings : none

import csv

def run(input, env):
    path = input.get('path')
    if not path:
        return {'error': 'No path provided'}

    totals = {}
    grand_total = 0.0

    try:
        with open(path, mode='r', newline='', encoding='utf-8') as csvfile:
            reader = csv.DictReader(csvfile)
            for row in reader:
                region = row['region']
                amount = float(row['amount'])

                totals[region] = totals.get(region, 0.0) + amount
                grand_total += amount
    except Exception as e:
        return {'error': str(e)}

    return {
        'totals': {k: round(v, 2) for k, v in totals.items()},
        'grand_total': round(grand_total, 2)
    }

4 · The artifact — a page you can send someone

The page blueprint renders a self-contained HTML file. That file is the right-hand pane from the reference UI, and in a notebook it renders inline: the agent's output is not a description of a dashboard, it is the dashboard.

In [8]

python

from IPython.display import HTML

page = store.create(
    "revenue_page", title="Revenue page", blueprint="page", overwrite=True
)

# `store.run` runs the app where the agent works. `run_app` on its own
# defaults to the app's own directory, and a relative path finds nothing there.
totals = store.run("revenue_by_region", {"path": "bookings.csv"})
rows = [
    {"Region": region, "Revenue": f"${amount:,.0f}"}
    for region, amount in sorted((totals.value or {}).get("totals", {}).items())
]

result = store.run("revenue_page", {"title": "Revenue by region · Q2",
                                    "rows": rows, "output": "revenue.html"})
print(result.value)

HTML(Path(result.value["path"]).read_text())

Out

{'path': '/tmp', 'bytes': 481, 'rows': 3}

rendered output

The artifact card

A file the run produced is not a path buried in tool output — it is a card, the way the reference UI shows Q2 Kickoff Brief · Doc · Click to open. Any tool that declares a path in its result metadata gets one, and the link is a plain file:// URL: no server, no viewer.

In [9]

python

from shipit_agent.models import AgentEvent
from shipit_agent.narrate import render_chat_html

# The event the runtime emits when a tool leaves a file behind.
made = AgentEvent(type="artifact_created", message="", payload={
    "path": result.value["path"],
    "title": "Revenue by region · Q2",
    "kind": "Page",
    "tool": "use_app",
})

HTML(render_chat_html(
    [AgentEvent(type="text_delta", message="",
                payload={"chunk": "Here is the page — it is yours to send."}),
     made,
     AgentEvent(type="run_completed", message="",
                payload={"output": "", "usage": {"total_tokens": 4200}})],
    title="Revenue analysis", model=MODEL,
))

Out

rendered output

Every tool that writes a file gets this for free. build_document producing a brief, build_artifact producing a workbook, an app writing a page — each one declares its path, and each one becomes a card.

In [10]

python

# Two artifacts, as a UI would show them side by side.
doc = WORKSPACE / "q2_brief.md"
doc.write_text("# Q2 Kickoff Brief\n\nRevenue is tracking to plan.\n")
sheet = WORKSPACE / "q2_workstreams.csv"
sheet.write_text("workstream,owner,status\nGateway,Dana Kim,On track\n")

cards = [
    AgentEvent(type="text_delta", message="", payload={
        "chunk": "I'll start with the brief, then pull the workstreams "
                 "into a tracker."}),
    AgentEvent(type="artifact_created", message="", payload={
        "path": str(doc), "title": "Q2 Kickoff Brief", "kind": "Doc",
        "tool": "build_document"}),
    AgentEvent(type="artifact_created", message="", payload={
        "path": str(sheet), "title": "Q2 Workstreams", "kind": "Sheet",
        "tool": "build_document"}),
    AgentEvent(type="run_completed", message="", payload={
        "output": "Both are ready.", "usage": {"total_tokens": 31905}}),
]
HTML(render_chat_html(cards, title="Q2 kickoff pack", model=MODEL))

Out

rendered output

6 · Run it again — no model, no tokens

This is the whole argument for building an app rather than answering a question. The analysis is a function now.

In [11]

python

# A different file, the same app. Nothing here calls an LLM.
(WORKSPACE / "q3.csv").write_text(
    "region,rep,amount,month\n"
    "EMEA,Dana Kim,190000,Jul\n"
    "AMER,Sam Osei,310000,Aug\n"
    "APAC,Maximo Guk,88000,Sep\n"
)

for path in ("bookings.csv", "q3.csv"):
    out = store.run("revenue_by_region", {"path": path})
    print(f"{path:<14} {out.value}")

Out

bookings.csv   {'totals': {'EMEA': 204000.0, 'AMER': 494000.0, 'APAC': 109000.0}, 'grand_total': 807000.0}

q3.csv         {'totals': {'EMEA': 190000.0, 'AMER': 310000.0, 'APAC': 88000.0}, 'grand_total': 588000.0}

7 · Progress narration — what it is doing, while it does it

progress_summaries=True puts a second model alongside the run. It sees the tools that were selected, the arguments they were given and what they returned — never the system prompt, never hidden reasoning — and writes one sentence before each step and one after.

It costs a real LLM call each time, which is why it is off by default. Give it a cheap decision_llm in production.

In [12]

python

narrated = Agent(
    llm=llm(), tools=[*READERS, *APPS], permissions=POLICY,
    auto_use_skills=False, max_iterations=8,
    progress_summaries=True,          # ← the flag
    # decision_llm=cheap_llm,         # ← and a cheaper model for it
)

narrated_events = []
for event in narrated.stream(
    "Use use_app on revenue_by_region with path 'q3.csv', then compare the "
    "regional totals to targets.csv and say which regions are behind."
):
    narrated_events.append(event)
    if event.type == "agent_decision":
        print(f"\n▸ {event.payload['summary']}")
    elif event.type == "agent_observation":
        print(f"  ← {event.payload['summary']}")
    elif event.type == "progress_summary_failed":
        print(f"  ! summary failed: {event.payload.get('error', '')[:120]}")

Out

▸ I'm searching for 'q3.csv' to confirm its location and ensure it is available before attempting to load it into the revenue_by_region app.

  ← The file 'q3.csv' was successfully located in the directory.

▸ The agent is searching for `targets.csv` to retrieve the regional targets needed for the comparison. Once found, it will compare those targets against the data from `q3.csv` to identify which regions are behind.

  ← The agent located the `targets.csv` file using the glob_files tool.

▸ I am reading the data in `q3.csv` to begin calculating the regional revenue totals for comparison.

  ← The file `revenue_by_region` was successfully read, showing regional revenue amounts for EMEA ($190,000), AMER ($310,000), and APAC ($88,000).

▸ Reading `targets.csv` to retrieve the regional targets needed for comparison with the revenue totals.

  ← The `read_file` tool successfully retrieved target values from `targets.csv`, showing targets of 200,000 for EMEA, 450,000 for AMER, and 150,000 for APAC.

▸ I'm searching for the `revenue_by_region` application blueprint to determine how to process the revenue data and compare it against the targets.

  ← I listed the available blueprints and identified that the `revenue_by_region` app already exists.

▸ I'm loading the revenue data from 'q3.csv' using the revenue_by_region app so that I can compare the regional amounts against the targets provided in 'targets.csv'.

  ← The application processed 'q3.csv' and calculated regional revenue totals: EMEA at 190,000.0, AMER at 310,000.0, and APAC at 88,000.0, resulting in a grand total of 588,000.0.

▸ The agent will compare the regional revenue totals from the `revenue_by_region` app against the targets in `targets.csv` to identify which specific regions are currently behind their goals.

The same run as a tree

Decisions and observations are rows of their own — narration about the run, kept distinct from the answer to the user.

In [13]

python

print(render_tree(narrated_events, model="gemma-4-26b"))

Out

Agent started
│
├─ Decision
│ I'm searching for 'q3.csv' to confirm its location and ensure it is available before attempting
│ to load it into the revenue_by_region app.
│
├─ Tool group: Searched for q3.csv
│ └─ glob_files completed 1ms
│
├─ Observed
│ The file 'q3.csv' was successfully located in the directory.
│
├─ Decision
│ The agent is searching for `targets.csv` to retrieve the regional targets needed for the
│ comparison. Once found, it will compare those targets against the data from `q3.csv` to
│ identify which regions are behind.
│
├─ Tool group: Searched for targets.csv
│ └─ glob_files completed 2ms
│
├─ Observed
│ The agent located the `targets.csv` file using the glob_files tool.
│
├─ Decision
│ I am reading the data in `q3.csv` to begin calculating the regional revenue totals for
│ comparison.
│
├─ Tool group: Read q3.csv
│ └─ read_file completed 1ms
│
├─ Observed
│ The file `revenue_by_region` was successfully read, showing regional revenue amounts for EMEA
│ ($190,000), AMER ($310,000), and APAC ($88,000).
│
├─ Decision
│ Reading `targets.csv` to retrieve the regional targets needed for comparison with the revenue
│ totals.
│
├─ Tool group: Read targets.csv
│ └─ read_file completed 2ms
… 46 more lines

8 · The panel, the tree, and the JSON

Three views of the same events. The panel is what a person reads; the timeline is what a frontend draws.

In [14]

python

from shipit_agent.narrate import render_chat_html

HTML(render_chat_html(narrated_events, model=MODEL,
                      title="Revenue analysis", output_limit=None))

Out

rendered output

In [15]

python

import json

from shipit_agent.narrate import timeline

for step in timeline(narrated_events):
    rest = {k: v for k, v in step.items() if k != "type"}
    print(f"{step['type']:<24} {json.dumps(rest, default=str)[:96]}")

Out

run_started              {"goal": "Use use_app on revenue_by_region with path 'q3.csv', then compare the regional totals 
agent_decision           {"content": "I'm searching for 'q3.csv' to confirm its location and ensure it is available befor
tool_group_started       {"group_id": "g1", "title": "Searching for q3.csv"}
tool_call_started        {"tool_call_id": "call_1_1", "group_id": "g1", "tool_name": "glob_files", "input": {"pattern": "
tool_call_completed      {"tool_call_id": "call_1_1", "group_id": "g1", "tool_name": "glob_files", "status": "completed",
tool_group_completed     {"group_id": "g1", "tool_calls": 1}
agent_observation        {"content": "The file 'q3.csv' was successfully located in the directory.", "next_action": "eval
agent_decision           {"content": "The agent is searching for `targets.csv` to retrieve the regional targets needed fo
tool_group_started       {"group_id": "g2", "title": "Searching for targets.csv"}
tool_call_started        {"tool_call_id": "call_2_1", "group_id": "g2", "tool_name": "glob_files", "input": {"pattern": "
tool_call_completed      {"tool_call_id": "call_2_1", "group_id": "g2", "tool_name": "glob_files", "status": "completed",
tool_group_completed     {"group_id": "g2", "tool_calls": 1}
agent_observation        {"content": "The agent located the `targets.csv` file using the glob_files tool.", "next_action"
agent_decision           {"content": "I am reading the data in `q3.csv` to begin calculating the regional revenue totals 
tool_group_started       {"group_id": "g3", "title": "Reading q3.csv"}
tool_call_started        {"tool_call_id": "call_3_1", "group_id": "g3", "tool_name": "read_file", "input": {"path": "q3.c
tool_call_completed      {"tool_call_id": "call_3_1", "group_id": "g3", "tool_name": "read_file", "status": "completed", 
tool_group_completed     {"group_id": "g3", "tool_calls": 1}
agent_observation        {"content": "The file `revenue_by_region` was successfully read, showing regional revenue amount
agent_decision           {"content": "Reading `targets.csv` to retrieve the regional targets needed for comparison with t
tool_group_started       {"group_id": "g4", "title": "Reading targets.csv"}
tool_call_started        {"tool_call_id": "call_4_1", "group_id": "g4", "tool_name": "read_file", "input": {"path": "targ
tool_call_completed      {"tool_call_id": "call_4_1", "group_id": "g4", "tool_name": "read_file", "status": "completed", 
tool_group_completed     {"group_id": "g4", "tool_calls": 1}
agent_observation        {"content": "The `read_file` tool successfully retrieved target values from `targets.csv`, showi
agent_decision           {"content": "I'm searching for the `revenue_by_region` application blueprint to determine how to
tool_group_started       {"group_id": "g5", "title": "Listing blueprints"}
tool_call_started        {"tool_call_id": "call_5_1", "group_id": "g5", "tool_name": "list_blueprints", "input": {}}
tool_call_completed      {"tool_call_id": "call_5_1", "group_id": "g5", "tool_name": "list_blueprints", "status": "comple
tool_group_completed     {"group_id": "g5", "tool_calls": 1}
agent_observation        {"content": "I listed the available blueprints and identified that the `revenue_by_region` app a
agent_decision           {"content": "I'm loading the revenue data from 'q3.csv' using the revenue_by_region app so that 
tool_group_started       {"group_id": "g6", "title": "Using the app revenue_by_region"}
tool_call_started        {"tool_call_id": "call_6_1", "group_id": "g6", "tool_name": "use_app", "input": {"app": "revenue
tool_call_completed      {"tool_call_id": "call_6_1", "group_id": "g6", "tool_name": "use_app", "status": "completed", "d
tool_group_completed     {"group_id": "g6", "tool_calls": 1}
agent_observation        {"content": "The application processed 'q3.csv' and calculated regional revenue totals: EMEA at 
agent_decision           {"content": "The agent will compare the regional revenue totals from the `revenue_by_region` ap
… output trimmed

Note agent_decision and agent_observation in that stream: with progress_summaries off they are absent, and a frontend simply draws fewer rows. Nothing else changes shape.

9 · Sub-agents the agent brings itself

Note what is not in the tools list below: sub_agent. Nobody added it. delegation=True guarantees it exists — built from this agent's own LLM and its read-only tools — and, when a task sizes up as several independent pieces, appends the instruction to the task where a small model acts on it.

Two examples: one where the work is obviously separable, and one where it is not. The second is the important one — a policy that delegates everything is just a slower agent.

In [16]

python

# Three regional files. The agent is given readers only — no sub_agent.
for region, rows in {
    "emea.csv": "rep,amount\nDana Kim,120000\nLuis Marin,84000\n",
    "amer.csv": "rep,amount\nSam Osei,240000\nPriya Raman,156000\n",
    "apac.csv": "rep,amount\nMaximo Guk,64000\nKenton Varda,45000\n",
}.items():
    (WORKSPACE / region).write_text(rows)

auto = Agent(
    llm=llm(),
    tools=READERS,                    # ← no sub_agent here
    permissions=PermissionEngine(
        allow=["read_file", "glob_files", "grep_files", "sub_agent"],
        default_decision=PermissionDecision.DENY),
    auto_use_skills=False, max_iterations=10,
    delegation=True,                  # ← the agent brings its own
)

print("tools the agent was given :", [t.name for t in READERS])
print("tools it actually has     :",
      sorted(t.name for t in auto._effective_tools("x") if hasattr(t, "name")))

Out

tools the agent was given : ['read_file', 'glob_files', 'grep_files']
tools it actually has     : ['glob_files', 'grep_files', 'read_file', 'sub_agent']

Example 1 — separable work, delegated without being asked

The prompt never says "sub-agent" or "delegate".

In [17]

python

wide = list(auto.stream(
    "Summarize emea.csv, amer.csv and apac.csv — for each one, the total "
    "amount and the top rep. Then say which region is largest."
))

parent = [e.payload["tool"] for e in wide if e.type == "tool_called"]
children = [(e.payload.get("agent"), (e.payload.get("inner") or {}).get("tool"))
            for e in wide
            if e.type == "sub_agent_event"
            and e.payload.get("inner_type") == "tool_called"]

print("parent calls :", parent)
print("delegations  :", parent.count("sub_agent"))
print("child calls  :", children)
print()
print("".join(e.payload["chunk"] for e in wide if e.type == "text_delta")[:600])

Out

parent calls : ['sub_agent', 'sub_agent', 'sub_agent', 'sub_agent', 'sub_agent', 'sub_agent', 'sub_agent', 'sub_agent']
delegations  : 8
child calls  : [('sub-agent', 'glob_files'), ('sub-agent', 'read_file'), ('sub-agent', 'glob_files'), ('sub-agent', 'read_file'), ('sub-agent', 'glob_files'), ('sub-agent', 'read_file'), ('sub-agent', 'read_file'), ('sub-agent', 'read_file'), ('sub-agent', 'glob_files'), ('sub-agent', 'read_file'), ('sub-agent', 'glob_files'), ('sub-agent', 'read_file'), ('sub-agent', 'glob_files'), ('sub-agent', 'read_file')]

- EMEA: Total 204000, Top Rep Dana Kim
- AMER: Total 396000, Top Rep Sam Osei
- APAC: Total 109000, Top Rep Maximo Guk

The largest region is **AMER**.

Read the two call lists together. The children opened the files; the parent's context holds three one-line answers instead of three files. That is the whole economic argument, and it is visible in the numbers rather than asserted.

In [18]

python

print(render_tree(wide, model="gemma-4-26b"))

Out

Agent started
│
├─ Tool group: Started > Summarize emea.csv: total amount and top rep. Provide ans…
│  └─ sub_agent                                     completed  4ms
│
├─ Delegated: > Summarize emea.csv: total amount and top rep. Provide answ
│  ├─ glob_files                                    completed
│  └─ read_file                                     completed
│
├─ Tool group: Delegated
│  └─ sub_agent                                     completed  0ms
│
├─ Tool group: Delegated > Summarize amer.csv: total amount and top rep. Provide ans…
│  └─ sub_agent                                     completed  2184ms
│
├─ Delegated: > Summarize amer.csv: total amount and top rep. Provide answ
│  ├─ glob_files                                    completed
│  └─ read_file                                     completed
│
├─ Tool group: Delegated > Summarize amer.csv: total amount and top rep. Provide ans…
│  └─ sub_agent                                     completed  9839ms
│
├─ Delegated: > Summarize amer.csv: total amount and top rep. Provide answ
│  ├─ glob_files                                    completed
│  ├─ read_file                                     completed
│  ├─ read_file                                     completed
│  └─ read_file                                     completed
│
├─ Tool group: Delegated > Summarize amer.csv: total amount and top rep. Provide ans…
│  └─ sub_agent                                     completed  2408ms
│
├─ Delegated: > Summarize amer.csv: total amount and top rep. Provide answ
│  ├─ glob_files                                    completed
│  └─ read_file                                     completed
│
├─ Tool group: Delegated
│  └─ sub_agent                                     completed  0ms
│
├─ Tool group: Delegated > Summarize apac.csv: total amount and top rep. Provide ans…
│  └─ sub_agent                                     completed  4124ms
… 20 more lines

Example 2 — one indivisible question, not delegated

Same agent, same flag. A single lookup has nothing to split, so the policy stays quiet and the agent just answers.

In [19]

python

from shipit_agent.delegation import DelegationPolicy, StructuralAssessor

narrow = list(auto.stream("What is the total amount in emea.csv?"))
narrow_calls = [e.payload["tool"] for e in narrow if e.type == "tool_called"]

print("parent calls :", narrow_calls)
print("delegations  :", narrow_calls.count("sub_agent"))
print()

# Why: the policy sizes the task before the model ever sees it.
policy = DelegationPolicy(assessor=StructuralAssessor())
for task in ["What is the total amount in emea.csv?",
             "Summarize emea.csv, amer.csv and apac.csv."]:
    advice = policy.assess(task)
    verdict = f"delegate ×{advice.items}" if advice else "do it yourself"
    print(f"{verdict:<18} {task}")
    for reason in advice.reasons:
        print(f"{'':<18} · {reason}")

Out

parent calls : ['glob_files', 'read_file']
delegations  : 0

do it yourself     What is the total amount in emea.csv?
delegate ×3        Summarize emea.csv, amer.csv and apac.csv.
                   · 3 concrete targets are named

By default the sizing is done by a model, not the word list above — one cheap cached question about the task. StructuralAssessor is the zero-cost fallback, and it counts structure only: enumerated lists, named targets, stated quantities. No keyword table to go stale in a language nobody tested.

10 · Connections — the agent asks, you answer

A missing connection is not an error the agent should work around. It asks, with a reason, and the run carries a connection_requested event a UI draws a card from — the same shape as the reference product's BigQuery — analytics.usage · Connect.

In [20]

python

from shipit_agent.integrations import InMemoryCredentialStore
from shipit_agent.tools.connections import ConnectionsTool


class Warehouse:
    """A connector with no credential yet — the interesting case."""

    name = "warehouse"
    description = "Query the analytics warehouse."
    prompt_instructions = ""
    credential_key = "warehouse"

    def schema(self):
        return {"type": "function", "function": {
            "name": self.name, "description": self.description,
            "parameters": {"type": "object",
                           "properties": {"query": {"type": "string"}},
                           "required": ["query"]}}}

    def run(self, context, **kwargs):
        from shipit_agent.tools.base import ToolOutput

        return ToolOutput(text="(never reached without a credential)")


connected = Agent(
    llm=llm(), tools=[ConnectionsTool(), Warehouse(), *READERS],
    credential_store=InMemoryCredentialStore(),
    auto_use_skills=False, max_iterations=6,
)

conn_events = list(connected.stream(
    "Check with the connections tool whether the warehouse is connected. "
    "If it is not, call connections with action='request', connection="
    "'warehouse' and a reason explaining what you need it for."
))

for event in conn_events:
    if event.type == "connection_requested":
        print("REQUEST:", event.payload)

The card

Whatever the model managed to say, the request is a structured event — and this is what a UI renders from it.

In [21]

python

from shipit_agent.models import AgentEvent

# If the live run produced one, show that; otherwise show the shape, so the
# card is visible either way (this cell never fabricates a run).
request = next(
    (e for e in conn_events if e.type == "connection_requested"),
    AgentEvent(type="connection_requested", message="", payload={
        "connection_id": "warehouse", "title": "Warehouse",
        "reason": "Read the bookings tables to total revenue by region.",
        "auth": "api_key", "tool": "connections"}),
)
HTML(render_chat_html([request], title="Connection request"))

Out

rendered output

Answering it

resolve() is the other half of the card. On accept with a credential, the credential is stored — so the next state check reads connected rather than asking you again for something you just gave it.

In [22]

python

from shipit_agent import ConnectionRegistry

store_ = InMemoryCredentialStore()
registry = ConnectionRegistry(credential_store=store_, tools=[Warehouse()])
registry.request("warehouse", "Read the bookings tables.")

print("before :", registry.render().strip())
print("pending:", [r.title for r in registry.pending_requests()])

registry.resolve("warehouse", accepted=True, credential="wh-key-123", by="you")

print()
print("after  :", registry.render().strip())
print("pending:", registry.pending_requests() or "none")

Out

before : · Warehouse — not connected
      tools: warehouse
      → Connect Warehouse.
pending: ['Warehouse']

after  : ✓ Warehouse — connected
      tools: warehouse
pending: none

11 · What an app may reach

An app is never more privileged than the agent that wrote it. It runs in a subprocess with no credentials, and its env carries only the bindings its manifest names — every call on them crossing back to the parent's permission engine.

Below: two bindings exist, one is wired, and the app can only see that one.

In [23]

python

peek = store.create("peek", title="Peek", overwrite=True, files={"app.py": """
def run(input, env):
    return {"visible": sorted(dir(env))}
"""})
store.bind("peek", source="ALLOWED")


class FakeBinding:
    methods = {"call": None}


def invoker(binding, method, kwargs):
    return "ok", {}


seen = store.run("peek", invoker=invoker,
                 bindings={"ALLOWED": FakeBinding(), "SECRET": FakeBinding()})
print("wired    :", store.get("peek").manifest.bindings)
print("app sees :", seen.value["visible"])
assert "SECRET" not in seen.value["visible"]
print("\nSECRET was never wired, so the app cannot name it — let alone call it.")

Out

wired    : {'ALLOWED': 'ALLOWED'}
app sees : ['ALLOWED']

SECRET was never wired, so the app cannot name it — let alone call it.

Where each piece lives

You want	Use
Something reusable	`create_app` → `use_app`
To limit what it reaches	`set_app_binding`
To call it without a model	`run_app(store.get(name), {...})`
Narration while it works	`Agent(progress_summaries=True, decision_llm=…)`
A card when something needs connecting	`connections` + `registry.resolve()`
A card for a file you made	declare its path in tool metadata
Sub-agents without adding one	`Agent(delegation=True)`
The page a person reads	`render_chat_html(events)`
The JSON your frontend draws	`timeline(events)`

Notebooks · Workspace, live UI and observability · 76

Apps, analysis, and the whole UI — end to end

The reference product's screenshot is a chat on the left and a thing on the right: a revenue dashboard, a spreadsheet, a workflow. The chat is where you ask; the thing is what you keep.

Open in Colab View on GitHub 23 code cells · 22 with saved output

Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.

1 · Setup

Same guard as the other notebooks: this repo on the path, and an assertion if some other shipit_agent wins.

In [ ]

python

import sys
from pathlib import Path


def repo_root(start: Path) -> Path:
    for candidate in [start, *start.parents]:
        if (candidate / "shipit_agent" / "__init__.py").exists():
            return candidate
    raise RuntimeError(f"Could not locate the shipit_agent repo from {start}")


REPO = repo_root(Path.cwd())
sys.path.insert(0, str(REPO))
for name in [n for n in sys.modules if n.startswith("shipit_agent")]:
    del sys.modules[name]

import shipit_agent

loaded = Path(shipit_agent.__file__).resolve().parent.parent
assert loaded == REPO.resolve(), f"wrong shipit_agent loaded: {loaded}"
print("repo    :", REPO)
print("version :", shipit_agent.__version__)

Out

repo    : ~/Documents/MYWORK/ai_developer/others/shipit_agent
version : 1.1.0

In [2]

python

import importlib.util

PROVIDER = Path(
    "~/Documents/MYWORK/AFTDRK/CACHE/DRK_CACHE_BACK"
    "/drk_cache/llm/bedrock_mantle_provider.py"
)
MODEL = "bedrock-mantle/google.gemma-4-26b-a4b"

spec = importlib.util.spec_from_file_location("bedrock_mantle_provider", PROVIDER)
module = importlib.util.module_from_spec(spec)
sys.modules["bedrock_mantle_provider"] = module
spec.loader.exec_module(module)
module.ensure_registered()

from shipit_agent.llms import LiteLLMChatLLM


def llm():
    return LiteLLMChatLLM(model=MODEL)


import boto3
print("identity:", boto3.client("sts").get_caller_identity()["Arn"])
print("model   :", MODEL)

Out

identity: arn:aws:iam::275210565507:user/0x99-bedrock
model   : bedrock-mantle/google.gemma-4-26b-a4b

The data

A quarter of bookings, as a CSV — the same shape as the "Revenue by region" panel in the reference UI, small enough to check by eye.

In [3]

python

import tempfile

WORKSPACE = Path(tempfile.mkdtemp(prefix="revenue_"))
(WORKSPACE / "bookings.csv").write_text(
    "region,rep,amount,month\n"
    "EMEA,Dana Kim,120000,Apr\n"
    "EMEA,Luis Marin,84000,May\n"
    "AMER,Sam Osei,240000,Apr\n"
    "AMER,Priya Raman,156000,May\n"
    "AMER,Dana Kim,98000,Jun\n"
    "APAC,Maximo Guk,64000,May\n"
    "APAC,Kenton Varda,45000,Jun\n"
)
(WORKSPACE / "targets.csv").write_text(
    "region,target\nEMEA,200000\nAMER,450000\nAPAC,150000\n"
)
print("workspace:", WORKSPACE)
print((WORKSPACE / "bookings.csv").read_text())

Out

workspace: /var/folders/bd/pq4lv0q52pv59m8pthkn60sc0000gn/T/revenue_eukt2id_
region,rep,amount,month
EMEA,Dana Kim,120000,Apr
EMEA,Luis Marin,84000,May
AMER,Sam Osei,240000,Apr
AMER,Priya Raman,156000,May
AMER,Dana Kim,98000,Jun
APAC,Maximo Guk,64000,May
APAC,Kenton Varda,45000,Jun

In [4]

python

from shipit_agent import Agent
from shipit_agent.builtins import get_builtin_tool_map
from shipit_agent.permissions import PermissionDecision, PermissionEngine

TOOLS = get_builtin_tool_map(llm=None, project_root=str(WORKSPACE))
READERS = [TOOLS[n] for n in ("read_file", "glob_files", "grep_files")]
APPS = [TOOLS[n] for n in
        ("list_blueprints", "create_app", "set_app_binding", "use_app")]

ALLOWED = [t.name for t in (*READERS, *APPS)]
POLICY = PermissionEngine(allow=ALLOWED, default_decision=PermissionDecision.DENY)

print("tools :", ALLOWED)

Out

tools : ['read_file', 'glob_files', 'grep_files', 'list_blueprints', 'create_app', 'set_app_binding', 'use_app']

2 · The four app tools

They ship as builtins, storing under <project>/.shipit/apps, so an app outlives the run that wrote it.

In [5]

python

from shipit_agent.tools.base import ToolContext

ctx = ToolContext(prompt="", state={})
print(TOOLS["list_blueprints"].run(ctx).text)

Out

Blueprints:
- `csv_summary` — CSV summary: Reads a CSV and counts rows, optionally grouped by a column.
- `dashboard` — Dashboard: Headline numbers and a bar chart from records — the thing you send someone who asked where revenue landed.
- `page` — HTML page: Renders a self-contained HTML page from records and writes it to disk — something you can send someone.
- `report` — Markdown report: Turns a list of records into a Markdown table.
- `sheet` — Spreadsheet: Records as a spreadsheet: column letters, row numbers, a frozen header, and cells you can flag.
- `workflow` — Workflow diagram: A pipeline as boxes and connectors, with the agent step marked and a live log underneath.

3 · The agent builds an analysis app

Not a script it throws away — an app, with a name, saved where you can run it again. The prompt says what it should do, not how to write it.

In [6]

python

from shipit_agent.narrate import render_tree

analyst = Agent(
    llm=llm(), tools=[*READERS, *APPS], permissions=POLICY,
    auto_use_skills=False, max_iterations=10,
)

BUILD = (
    "Create an app named revenue_by_region. Write its app.py yourself: "
    "run(input, env) must read the CSV at input['path'], total the `amount` "
    "column per `region`, and return {'totals': {...}, 'grand_total': N}. "
    "Then use_app it on path 'bookings.csv' and tell me the totals."
)

build_events = list(analyst.stream(BUILD))
print(render_tree(build_events, model="gemma-4-26b", detail=True, output_lines=10))

Out

Agent started
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed
│     ↳ "name"='revenue_by_region', "title"='Revenue by Region App', "description"='Calculates total revenue per region from a CSV file.', "code="<code>import csv<ol><l
│       Invalid app name '': use lower snake case, 2-41 characters, starting with a letter.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed
│     ↳ "name"='revenue_by_region', "title"='Revenue by Region App', "description"='Calculates total revenue per region from a CSV file.', "code="<code>import csv\\n\\n
│       Invalid app name '': use lower snake case, 2-41 characters, starting with a letter.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  0ms
│     ↳ "name"='revenue_by_region', "title"='Revenue by Region App', "code="many users often fail to provide the python code in a way that is correctly interpreted if t
│       Invalid app name '': use lower snake case, 2-41 characters, starting with a letter.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  0ms
│     ↳ "name"='revenue_by_region', "title"='Revenue by Region App', "code="bold(import csv\\n\\ndef run(input, env):\\n    path = input.get('path')\\n    if not path:\
│       Invalid app name '': use lower snake case, 2-41 characters, starting with a letter.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  0ms
│     ↳ "name"='revenue_by_region', "title"='Revenue by Region App', "code="``import csv\\n\\ndef run(input, env):\\n    path = input.get('path')\\n    if not path:\\n 
│       Invalid app name '': use lower snake case, 2-41 characters, starting with a letter.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  1ms
│     ↳ name='revenue_by_region', title='Revenue by Region App', description='Calculates total revenue per region from a CSV file.', code:="import csv\n\ndef run(input,
│       An app needs app.py defining `run(input, env)` — pass it in files, or start from a blueprint.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  3ms
│     ↳ name='revenue_by_region', title='Revenue by Region App', code:="import csv\n\ndef run(input, env):\n    path = input.get('path')\n    if not path:\n        retu
│       An app needs app.py defining `run(input, env)` — pass it in files, or start from a blueprint.
│
├─ Tool group: Built the app revenue_by_region
│  └─ create_app                                    completed  4ms
│     ↳ name='revenue_by_region', title='Revenue by Region App', code="import csv\n\ndef run(input, env):\n    path = input.get('path')\n    if not path:\n        retur
… 27 more lines

What it wrote

The app is a file on disk. Read it — this is the thing you keep.

In [7]

python

from shipit_agent.apps import AppStore, run_app

store = AppStore(WORKSPACE / ".shipit" / "apps", workdir=WORKSPACE)
for app in store.list():
    print(app.describe())
    print()
    print(app.entrypoint.read_text())

Out

Revenue by Region App (`revenue_by_region`)
files    : app.json, app.py
bindings : none

import csv

def run(input, env):
    path = input.get('path')
    if not path:
        return {'error': 'No path provided'}

    totals = {}
    grand_total = 0.0

    try:
        with open(path, mode='r', newline='', encoding='utf-8') as csvfile:
            reader = csv.DictReader(csvfile)
            for row in reader:
                region = row['region']
                amount = float(row['amount'])

                totals[region] = totals.get(region, 0.0) + amount
                grand_total += amount
    except Exception as e:
        return {'error': str(e)}

    return {
        'totals': {k: round(v, 2) for k, v in totals.items()},
        'grand_total': round(grand_total, 2)
    }

4 · The artifact — a page you can send someone

In [8]

python

from IPython.display import HTML

page = store.create(
    "revenue_page", title="Revenue page", blueprint="page", overwrite=True
)

# `store.run` runs the app where the agent works. `run_app` on its own
# defaults to the app's own directory, and a relative path finds nothing there.
totals = store.run("revenue_by_region", {"path": "bookings.csv"})
rows = [
    {"Region": region, "Revenue": f"${amount:,.0f}"}
    for region, amount in sorted((totals.value or {}).get("totals", {}).items())
]

result = store.run("revenue_page", {"title": "Revenue by region · Q2",
                                    "rows": rows, "output": "revenue.html"})
print(result.value)

HTML(Path(result.value["path"]).read_text())

Out

{'path': '/tmp', 'bytes': 481, 'rows': 3}

rendered output

The artifact card

In [9]

python

from shipit_agent.models import AgentEvent
from shipit_agent.narrate import render_chat_html

# The event the runtime emits when a tool leaves a file behind.
made = AgentEvent(type="artifact_created", message="", payload={
    "path": result.value["path"],
    "title": "Revenue by region · Q2",
    "kind": "Page",
    "tool": "use_app",
})

HTML(render_chat_html(
    [AgentEvent(type="text_delta", message="",
                payload={"chunk": "Here is the page — it is yours to send."}),
     made,
     AgentEvent(type="run_completed", message="",
                payload={"output": "", "usage": {"total_tokens": 4200}})],
    title="Revenue analysis", model=MODEL,
))

Out

rendered output

In [10]

python

# Two artifacts, as a UI would show them side by side.
doc = WORKSPACE / "q2_brief.md"
doc.write_text("# Q2 Kickoff Brief\n\nRevenue is tracking to plan.\n")
sheet = WORKSPACE / "q2_workstreams.csv"
sheet.write_text("workstream,owner,status\nGateway,Dana Kim,On track\n")

cards = [
    AgentEvent(type="text_delta", message="", payload={
        "chunk": "I'll start with the brief, then pull the workstreams "
                 "into a tracker."}),
    AgentEvent(type="artifact_created", message="", payload={
        "path": str(doc), "title": "Q2 Kickoff Brief", "kind": "Doc",
        "tool": "build_document"}),
    AgentEvent(type="artifact_created", message="", payload={
        "path": str(sheet), "title": "Q2 Workstreams", "kind": "Sheet",
        "tool": "build_document"}),
    AgentEvent(type="run_completed", message="", payload={
        "output": "Both are ready.", "usage": {"total_tokens": 31905}}),
]
HTML(render_chat_html(cards, title="Q2 kickoff pack", model=MODEL))

Out

rendered output

6 · Run it again — no model, no tokens

This is the whole argument for building an app rather than answering a question. The analysis is a function now.

In [11]

python

# A different file, the same app. Nothing here calls an LLM.
(WORKSPACE / "q3.csv").write_text(
    "region,rep,amount,month\n"
    "EMEA,Dana Kim,190000,Jul\n"
    "AMER,Sam Osei,310000,Aug\n"
    "APAC,Maximo Guk,88000,Sep\n"
)

for path in ("bookings.csv", "q3.csv"):
    out = store.run("revenue_by_region", {"path": path})
    print(f"{path:<14} {out.value}")

Out

bookings.csv   {'totals': {'EMEA': 204000.0, 'AMER': 494000.0, 'APAC': 109000.0}, 'grand_total': 807000.0}

q3.csv         {'totals': {'EMEA': 190000.0, 'AMER': 310000.0, 'APAC': 88000.0}, 'grand_total': 588000.0}

7 · Progress narration — what it is doing, while it does it

It costs a real LLM call each time, which is why it is off by default. Give it a cheap decision_llm in production.

In [12]

python

narrated = Agent(
    llm=llm(), tools=[*READERS, *APPS], permissions=POLICY,
    auto_use_skills=False, max_iterations=8,
    progress_summaries=True,          # ← the flag
    # decision_llm=cheap_llm,         # ← and a cheaper model for it
)

narrated_events = []
for event in narrated.stream(
    "Use use_app on revenue_by_region with path 'q3.csv', then compare the "
    "regional totals to targets.csv and say which regions are behind."
):
    narrated_events.append(event)
    if event.type == "agent_decision":
        print(f"\n▸ {event.payload['summary']}")
    elif event.type == "agent_observation":
        print(f"  ← {event.payload['summary']}")
    elif event.type == "progress_summary_failed":
        print(f"  ! summary failed: {event.payload.get('error', '')[:120]}")

Out

▸ I'm searching for 'q3.csv' to confirm its location and ensure it is available before attempting to load it into the revenue_by_region app.

  ← The file 'q3.csv' was successfully located in the directory.

▸ The agent is searching for `targets.csv` to retrieve the regional targets needed for the comparison. Once found, it will compare those targets against the data from `q3.csv` to identify which regions are behind.

  ← The agent located the `targets.csv` file using the glob_files tool.

▸ I am reading the data in `q3.csv` to begin calculating the regional revenue totals for comparison.

  ← The file `revenue_by_region` was successfully read, showing regional revenue amounts for EMEA ($190,000), AMER ($310,000), and APAC ($88,000).

▸ Reading `targets.csv` to retrieve the regional targets needed for comparison with the revenue totals.

  ← The `read_file` tool successfully retrieved target values from `targets.csv`, showing targets of 200,000 for EMEA, 450,000 for AMER, and 150,000 for APAC.

▸ I'm searching for the `revenue_by_region` application blueprint to determine how to process the revenue data and compare it against the targets.

  ← I listed the available blueprints and identified that the `revenue_by_region` app already exists.

▸ I'm loading the revenue data from 'q3.csv' using the revenue_by_region app so that I can compare the regional amounts against the targets provided in 'targets.csv'.

  ← The application processed 'q3.csv' and calculated regional revenue totals: EMEA at 190,000.0, AMER at 310,000.0, and APAC at 88,000.0, resulting in a grand total of 588,000.0.

▸ The agent will compare the regional revenue totals from the `revenue_by_region` app against the targets in `targets.csv` to identify which specific regions are currently behind their goals.

The same run as a tree

Decisions and observations are rows of their own — narration about the run, kept distinct from the answer to the user.

In [13]

python

print(render_tree(narrated_events, model="gemma-4-26b"))

Out

8 · The panel, the tree, and the JSON

Three views of the same events. The panel is what a person reads; the timeline is what a frontend draws.

In [14]

python

from shipit_agent.narrate import render_chat_html

HTML(render_chat_html(narrated_events, model=MODEL,
                      title="Revenue analysis", output_limit=None))

Out

rendered output

In [15]

python

import json

from shipit_agent.narrate import timeline

for step in timeline(narrated_events):
    rest = {k: v for k, v in step.items() if k != "type"}
    print(f"{step['type']:<24} {json.dumps(rest, default=str)[:96]}")

Out

run_started              {"goal": "Use use_app on revenue_by_region with path 'q3.csv', then compare the regional totals 
agent_decision           {"content": "I'm searching for 'q3.csv' to confirm its location and ensure it is available befor
tool_group_started       {"group_id": "g1", "title": "Searching for q3.csv"}
tool_call_started        {"tool_call_id": "call_1_1", "group_id": "g1", "tool_name": "glob_files", "input": {"pattern": "
tool_call_completed      {"tool_call_id": "call_1_1", "group_id": "g1", "tool_name": "glob_files", "status": "completed",
tool_group_completed     {"group_id": "g1", "tool_calls": 1}
agent_observation        {"content": "The file 'q3.csv' was successfully located in the directory.", "next_action": "eval
agent_decision           {"content": "The agent is searching for `targets.csv` to retrieve the regional targets needed fo
tool_group_started       {"group_id": "g2", "title": "Searching for targets.csv"}
tool_call_started        {"tool_call_id": "call_2_1", "group_id": "g2", "tool_name": "glob_files", "input": {"pattern": "
tool_call_completed      {"tool_call_id": "call_2_1", "group_id": "g2", "tool_name": "glob_files", "status": "completed",
tool_group_completed     {"group_id": "g2", "tool_calls": 1}
agent_observation        {"content": "The agent located the `targets.csv` file using the glob_files tool.", "next_action"
agent_decision           {"content": "I am reading the data in `q3.csv` to begin calculating the regional revenue totals 
tool_group_started       {"group_id": "g3", "title": "Reading q3.csv"}
tool_call_started        {"tool_call_id": "call_3_1", "group_id": "g3", "tool_name": "read_file", "input": {"path": "q3.c
tool_call_completed      {"tool_call_id": "call_3_1", "group_id": "g3", "tool_name": "read_file", "status": "completed", 
tool_group_completed     {"group_id": "g3", "tool_calls": 1}
agent_observation        {"content": "The file `revenue_by_region` was successfully read, showing regional revenue amount
agent_decision           {"content": "Reading `targets.csv` to retrieve the regional targets needed for comparison with t
tool_group_started       {"group_id": "g4", "title": "Reading targets.csv"}
tool_call_started        {"tool_call_id": "call_4_1", "group_id": "g4", "tool_name": "read_file", "input": {"path": "targ
tool_call_completed      {"tool_call_id": "call_4_1", "group_id": "g4", "tool_name": "read_file", "status": "completed", 
tool_group_completed     {"group_id": "g4", "tool_calls": 1}
agent_observation        {"content": "The `read_file` tool successfully retrieved target values from `targets.csv`, showi
agent_decision           {"content": "I'm searching for the `revenue_by_region` application blueprint to determine how to
tool_group_started       {"group_id": "g5", "title": "Listing blueprints"}
tool_call_started        {"tool_call_id": "call_5_1", "group_id": "g5", "tool_name": "list_blueprints", "input": {}}
tool_call_completed      {"tool_call_id": "call_5_1", "group_id": "g5", "tool_name": "list_blueprints", "status": "comple
tool_group_completed     {"group_id": "g5", "tool_calls": 1}
agent_observation        {"content": "I listed the available blueprints and identified that the `revenue_by_region` app a
agent_decision           {"content": "I'm loading the revenue data from 'q3.csv' using the revenue_by_region app so that 
tool_group_started       {"group_id": "g6", "title": "Using the app revenue_by_region"}
tool_call_started        {"tool_call_id": "call_6_1", "group_id": "g6", "tool_name": "use_app", "input": {"app": "revenue
tool_call_completed      {"tool_call_id": "call_6_1", "group_id": "g6", "tool_name": "use_app", "status": "completed", "d
tool_group_completed     {"group_id": "g6", "tool_calls": 1}
agent_observation        {"content": "The application processed 'q3.csv' and calculated regional revenue totals: EMEA at 
agent_decision           {"content": "The agent will compare the regional revenue totals from the `revenue_by_region` ap
… output trimmed

Note agent_decision and agent_observation in that stream: with progress_summaries off they are absent, and a frontend simply draws fewer rows. Nothing else changes shape.

9 · Sub-agents the agent brings itself

Two examples: one where the work is obviously separable, and one where it is not. The second is the important one — a policy that delegates everything is just a slower agent.

In [16]

python

# Three regional files. The agent is given readers only — no sub_agent.
for region, rows in {
    "emea.csv": "rep,amount\nDana Kim,120000\nLuis Marin,84000\n",
    "amer.csv": "rep,amount\nSam Osei,240000\nPriya Raman,156000\n",
    "apac.csv": "rep,amount\nMaximo Guk,64000\nKenton Varda,45000\n",
}.items():
    (WORKSPACE / region).write_text(rows)

auto = Agent(
    llm=llm(),
    tools=READERS,                    # ← no sub_agent here
    permissions=PermissionEngine(
        allow=["read_file", "glob_files", "grep_files", "sub_agent"],
        default_decision=PermissionDecision.DENY),
    auto_use_skills=False, max_iterations=10,
    delegation=True,                  # ← the agent brings its own
)

print("tools the agent was given :", [t.name for t in READERS])
print("tools it actually has     :",
      sorted(t.name for t in auto._effective_tools("x") if hasattr(t, "name")))

Out

tools the agent was given : ['read_file', 'glob_files', 'grep_files']
tools it actually has     : ['glob_files', 'grep_files', 'read_file', 'sub_agent']

Example 1 — separable work, delegated without being asked

The prompt never says "sub-agent" or "delegate".

In [17]

python

wide = list(auto.stream(
    "Summarize emea.csv, amer.csv and apac.csv — for each one, the total "
    "amount and the top rep. Then say which region is largest."
))

parent = [e.payload["tool"] for e in wide if e.type == "tool_called"]
children = [(e.payload.get("agent"), (e.payload.get("inner") or {}).get("tool"))
            for e in wide
            if e.type == "sub_agent_event"
            and e.payload.get("inner_type") == "tool_called"]

print("parent calls :", parent)
print("delegations  :", parent.count("sub_agent"))
print("child calls  :", children)
print()
print("".join(e.payload["chunk"] for e in wide if e.type == "text_delta")[:600])

Out

parent calls : ['sub_agent', 'sub_agent', 'sub_agent', 'sub_agent', 'sub_agent', 'sub_agent', 'sub_agent', 'sub_agent']
delegations  : 8
child calls  : [('sub-agent', 'glob_files'), ('sub-agent', 'read_file'), ('sub-agent', 'glob_files'), ('sub-agent', 'read_file'), ('sub-agent', 'glob_files'), ('sub-agent', 'read_file'), ('sub-agent', 'read_file'), ('sub-agent', 'read_file'), ('sub-agent', 'glob_files'), ('sub-agent', 'read_file'), ('sub-agent', 'glob_files'), ('sub-agent', 'read_file'), ('sub-agent', 'glob_files'), ('sub-agent', 'read_file')]

- EMEA: Total 204000, Top Rep Dana Kim
- AMER: Total 396000, Top Rep Sam Osei
- APAC: Total 109000, Top Rep Maximo Guk

The largest region is **AMER**.

In [18]

python

print(render_tree(wide, model="gemma-4-26b"))

Out

Agent started
│
├─ Tool group: Started > Summarize emea.csv: total amount and top rep. Provide ans…
│  └─ sub_agent                                     completed  4ms
│
├─ Delegated: > Summarize emea.csv: total amount and top rep. Provide answ
│  ├─ glob_files                                    completed
│  └─ read_file                                     completed
│
├─ Tool group: Delegated
│  └─ sub_agent                                     completed  0ms
│
├─ Tool group: Delegated > Summarize amer.csv: total amount and top rep. Provide ans…
│  └─ sub_agent                                     completed  2184ms
│
├─ Delegated: > Summarize amer.csv: total amount and top rep. Provide answ
│  ├─ glob_files                                    completed
│  └─ read_file                                     completed
│
├─ Tool group: Delegated > Summarize amer.csv: total amount and top rep. Provide ans…
│  └─ sub_agent                                     completed  9839ms
│
├─ Delegated: > Summarize amer.csv: total amount and top rep. Provide answ
│  ├─ glob_files                                    completed
│  ├─ read_file                                     completed
│  ├─ read_file                                     completed
│  └─ read_file                                     completed
│
├─ Tool group: Delegated > Summarize amer.csv: total amount and top rep. Provide ans…
│  └─ sub_agent                                     completed  2408ms
│
├─ Delegated: > Summarize amer.csv: total amount and top rep. Provide answ
│  ├─ glob_files                                    completed
│  └─ read_file                                     completed
│
├─ Tool group: Delegated
│  └─ sub_agent                                     completed  0ms
│
├─ Tool group: Delegated > Summarize apac.csv: total amount and top rep. Provide ans…
│  └─ sub_agent                                     completed  4124ms
… 20 more lines

Example 2 — one indivisible question, not delegated

Same agent, same flag. A single lookup has nothing to split, so the policy stays quiet and the agent just answers.

In [19]

python

from shipit_agent.delegation import DelegationPolicy, StructuralAssessor

narrow = list(auto.stream("What is the total amount in emea.csv?"))
narrow_calls = [e.payload["tool"] for e in narrow if e.type == "tool_called"]

print("parent calls :", narrow_calls)
print("delegations  :", narrow_calls.count("sub_agent"))
print()

# Why: the policy sizes the task before the model ever sees it.
policy = DelegationPolicy(assessor=StructuralAssessor())
for task in ["What is the total amount in emea.csv?",
             "Summarize emea.csv, amer.csv and apac.csv."]:
    advice = policy.assess(task)
    verdict = f"delegate ×{advice.items}" if advice else "do it yourself"
    print(f"{verdict:<18} {task}")
    for reason in advice.reasons:
        print(f"{'':<18} · {reason}")

Out

parent calls : ['glob_files', 'read_file']
delegations  : 0

do it yourself     What is the total amount in emea.csv?
delegate ×3        Summarize emea.csv, amer.csv and apac.csv.
                   · 3 concrete targets are named

10 · Connections — the agent asks, you answer

In [20]

python

from shipit_agent.integrations import InMemoryCredentialStore
from shipit_agent.tools.connections import ConnectionsTool


class Warehouse:
    """A connector with no credential yet — the interesting case."""

    name = "warehouse"
    description = "Query the analytics warehouse."
    prompt_instructions = ""
    credential_key = "warehouse"

    def schema(self):
        return {"type": "function", "function": {
            "name": self.name, "description": self.description,
            "parameters": {"type": "object",
                           "properties": {"query": {"type": "string"}},
                           "required": ["query"]}}}

    def run(self, context, **kwargs):
        from shipit_agent.tools.base import ToolOutput

        return ToolOutput(text="(never reached without a credential)")


connected = Agent(
    llm=llm(), tools=[ConnectionsTool(), Warehouse(), *READERS],
    credential_store=InMemoryCredentialStore(),
    auto_use_skills=False, max_iterations=6,
)

conn_events = list(connected.stream(
    "Check with the connections tool whether the warehouse is connected. "
    "If it is not, call connections with action='request', connection="
    "'warehouse' and a reason explaining what you need it for."
))

for event in conn_events:
    if event.type == "connection_requested":
        print("REQUEST:", event.payload)

The card

Whatever the model managed to say, the request is a structured event — and this is what a UI renders from it.

In [21]

python

from shipit_agent.models import AgentEvent

# If the live run produced one, show that; otherwise show the shape, so the
# card is visible either way (this cell never fabricates a run).
request = next(
    (e for e in conn_events if e.type == "connection_requested"),
    AgentEvent(type="connection_requested", message="", payload={
        "connection_id": "warehouse", "title": "Warehouse",
        "reason": "Read the bookings tables to total revenue by region.",
        "auth": "api_key", "tool": "connections"}),
)
HTML(render_chat_html([request], title="Connection request"))

Out

rendered output

Answering it

In [22]

python

from shipit_agent import ConnectionRegistry

store_ = InMemoryCredentialStore()
registry = ConnectionRegistry(credential_store=store_, tools=[Warehouse()])
registry.request("warehouse", "Read the bookings tables.")

print("before :", registry.render().strip())
print("pending:", [r.title for r in registry.pending_requests()])

registry.resolve("warehouse", accepted=True, credential="wh-key-123", by="you")

print()
print("after  :", registry.render().strip())
print("pending:", registry.pending_requests() or "none")

Out

before : · Warehouse — not connected
      tools: warehouse
      → Connect Warehouse.
pending: ['Warehouse']

after  : ✓ Warehouse — connected
      tools: warehouse
pending: none

11 · What an app may reach

Below: two bindings exist, one is wired, and the app can only see that one.

In [23]

python

peek = store.create("peek", title="Peek", overwrite=True, files={"app.py": """
def run(input, env):
    return {"visible": sorted(dir(env))}
"""})
store.bind("peek", source="ALLOWED")


class FakeBinding:
    methods = {"call": None}


def invoker(binding, method, kwargs):
    return "ok", {}


seen = store.run("peek", invoker=invoker,
                 bindings={"ALLOWED": FakeBinding(), "SECRET": FakeBinding()})
print("wired    :", store.get("peek").manifest.bindings)
print("app sees :", seen.value["visible"])
assert "SECRET" not in seen.value["visible"]
print("\nSECRET was never wired, so the app cannot name it — let alone call it.")

Out

wired    : {'ALLOWED': 'ALLOWED'}
app sees : ['ALLOWED']

SECRET was never wired, so the app cannot name it — let alone call it.

Where each piece lives

You want	Use
Something reusable	`create_app` → `use_app`
To limit what it reaches	`set_app_binding`
To call it without a model	`run_app(store.get(name), {...})`
Narration while it works	`Agent(progress_summaries=True, decision_llm=…)`
A card when something needs connecting	`connections` + `registry.resolve()`
A card for a file you made	declare its path in tool metadata
Sub-agents without adding one	`Agent(delegation=True)`
The page a person reads	`render_chat_html(events)`
The JSON your frontend draws	`timeline(events)`