Stage the Tools: A Weekend Dry-Run Bus for Agent Side Projects

Weekend agent side projects usually fail at the tool boundary, not inside the prompt. A model proposes a call, the host runs it, and a missing key or a second retry writes to the wrong path. A dry-run bus records the intended call, checks a tiny contract, and forwards only after that check passes.

This article is a scoped weekend build log for that bus. The working demo is one Python module, a JSONL ticket log, and a short test file. Distributed queues, vendor lock-in, and production tracing were left on the floor.

The problem the bus is for

Side-project agents tend to grow a single run_tool(name, args) function. That function talks to the filesystem, the network, or a shell. It is convenient. It is also how a sloppy loop mutates local state before anyone reads the trace.

Tool-calling tutorials often jump straight to a live API. That is the wrong first step for a Saturday project. The first step is a ticket. Every proposed call becomes a row that can be inspected, rejected, or released.

A dry-run bus sits in front of execution. Live model traffic stays optional until the tickets look honest.

Weekend scope that survived the cut

The build kept three jobs. Everything else was treated as next-month work.

Kept

1. A JSONL ticket log on disk, one object per line.
2. A per-tool contract for required keys and simple JSON types.
3. A live adapter that stays off unless an environment flag is set.

Cut on purpose

• Redis, Kafka, or any networked queue.
• OpenTelemetry exporters and flame graphs.
• Prompt assembly, memory, and retrieval.
• Automatic dollar forecasts or vendor price tables.
• Multi-user auth.

The goal is a demo that runs before Sunday night. A platform can wait.

Artifact: a ticketed tool bus

The following module is a proposed weekend sketch. It is labeled as unexecuted design until the tests below are run on a local machine. Save it as toolbus.py.

from __future__ import annotations

import json
import os
import time
import uuid
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Callable, Mapping

Json = dict[str, Any]
LiveFn = Callable[[str, Json], Json]

CONTRACTS: dict[str, dict[str, type]] = {
    "read_file": {"path": str},
    "write_file": {"path": str, "contents": str},
    "http_get": {"url": str},
}

FORBIDDEN_KEYS = {"api_key", "token", "password", "authorization"}


@dataclass
class Ticket:
    ticket_id: str
    tool: str
    args: Json
    status: str
    reason: str = ""
    ts: float = field(default_factory=time.time)

    def row(self) -> Json:
        return asdict(self)


class ContractError(ValueError):
    pass


def validate(tool: str, args: Mapping[str, Any]) -> None:
    spec = CONTRACTS.get(tool)
    if spec is None:
        raise ContractError(f"unknown tool: {tool}")
    missing = [k for k in spec if k not in args]
    if missing:
        raise ContractError(f"missing keys: {missing}")
    for key, expected in spec.items():
        if not isinstance(args[key], expected):
            raise ContractError(f"{key} expected {expected.__name__}")
    leaked = FORBIDDEN_KEYS.intersection(args):
    if leaked:
        raise ContractError(f"forbidden keys: {sorted(leaked)}")
    if tool == "http_get":
        url = str(args["url"])
        if not url.startswith(("https://", "http://127.0.0.1", "http://localhost")):
            raise ContractError("url scheme not allowed")
    if tool in {"read_file", "write_file"}:
        path = Path(str(args["path"])).resolve()
        root = Path.cwd().resolve()
        if root not in path.parents and path != root:
            raise ContractError("path escapes workspace")


class ToolBus:
    def __init__(self, ledger: Path, live: LiveFn | None = None) -> None:
        self.ledger = ledger
        self.live = live
        self.ledger.parent.mkdir(parents=True, exist_ok=True)
        self.ledger.touch(exist_ok=True)

    def _append(self, ticket: Ticket) -> None:
        with self.ledger.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(ticket.row(), ensure_ascii=True) + "\n")

    def stage(self, tool: str, args: Json, *, execute: bool = False) -> Ticket:
        ticket_id = uuid.uuid4().hex[:12]
        try:
            validate(tool, args)
        except ContractError as exc:
            ticket = Ticket(ticket_id, tool, dict(args), "rejected", str(exc))
            self._append(ticket)
            return ticket

        if not execute:
            ticket = Ticket(ticket_id, tool, dict(args), "staged", "dry-run")
            self._append(ticket)
            return ticket

        if self.live is None:
            ticket = Ticket(ticket_id, tool, dict(args), "blocked", "no live adapter")
            self._append(ticket)
            return ticket

        result = self.live(tool, dict(args))
        ticket = Ticket(
            ticket_id,
            tool,
            {"args": dict(args), "result_keys": sorted(result)},
            "executed",
            "live",
        )
        self._append(ticket)
        return ticket

    def pending(self) -> list[Json]:
        rows: list[Json] = []
        for line in self.ledger.read_text(encoding="utf-8").splitlines():
            if not line.strip():
                continue
            row = json.loads(line)
            if row.get("status") == "staged":
                rows.append(row)
        return rows

The bus never executes on stage() unless execute=True. That is the whole weekend trick. Staging is cheap. Execution is a separate, explicit act.

A live adapter can be a local function. It can also be a small HTTP client pointed at an environment URL. The bus does not care. The ledger does.

import json
import urllib.request


def http_live(tool: str, args: dict) -> dict:
    url = os.environ["TOOLBUS_LIVE_URL"]
    payload = json.dumps({"tool": tool, "args": args}).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        return json.loads(resp.read().decode("utf-8"))

Leave TOOLBUS_LIVE_URL unset during the first evening. The dry-run path still writes tickets. That is enough to review a loop without spending a model call.

Working demo

Create a scratch folder and install nothing beyond the Python standard library.

mkdir -p /tmp/toolbus-demo && cd /tmp/toolbus-demo
cat > toolbus.py <<'PY'
# paste the module above
PY

A tiny driver shows the three ticket states: rejected, staged, and blocked.

# save as demo.py
from pathlib import Path
from toolbus import ToolBus

bus = ToolBus(Path("tickets.jsonl"))

print(bus.stage("wipe_disk", {"path": "/"}).status)
print(bus.stage("read_file", {"path": "README.md"}).status)
print(bus.stage("read_file", {"path": "README.md"}, execute=True).status)
print(len(bus.pending()))

printf '# demo\n' > README.md
python demo.py
cat tickets.jsonl

Expected shape of the log, not a measured benchmark:

• wipe_disk lands as rejected with unknown tool.
• The first read_file lands as staged.
• The second read_file lands as blocked because no live adapter was wired.
• pending() returns the staged row only.

That output is the demo. A reader can screenshot the JSONL file and stop. The live lane is optional homework.

Tests that prove the cut is honest

Save the following as test_toolbus.py. The tests pin behavior, not performance.

from pathlib import Path

from toolbus import ToolBus


def test_unknown_tool_is_rejected(tmp_path: Path) -> None:
    bus = ToolBus(tmp_path / "t.jsonl")
    ticket = bus.stage("shell", {"cmd": "ls"})
    assert ticket.status == "rejected"


def test_path_escape_is_rejected(tmp_path: Path) -> None:
    bus = ToolBus(tmp_path / "t.jsonl")
    ticket = bus.stage("write_file", {"path": "../out.txt", "contents": "x"})
    assert ticket.status == "rejected"


def test_dry_run_does_not_call_live(tmp_path: Path) -> None:
    calls: list[str] = []

    def live(tool: str, args: dict) -> dict:
        calls.append(tool)
        return {"ok": True}

    bus = ToolBus(tmp_path / "t.jsonl", live=live)
    ticket = bus.stage("http_get", {"url": "http://127.0.0.1:9/health"})
    assert ticket.status == "staged"
    assert calls == []


def test_execute_uses_live_adapter(tmp_path: Path) -> None:
    def live(tool: str, args: dict) -> dict:
        return {"status": 200, "tool": tool}

    bus = ToolBus(tmp_path / "t.jsonl", live=live)
    ticket = bus.stage(
        "http_get",
        {"url": "http://127.0.0.1:9/health"},
        execute=True,
    )
    assert ticket.status == "executed"

pip install pytest
pytest -q test_toolbus.py

If those four tests pass, the weekend scope is done. Extra assertions about latency or token counts were not part of the cut.

Decision table for dry-run versus live

Use the table as a gate, not as a slogan.

Signal on the ticket	Action	Why
Unknown tool name	Reject	The allow-set is the contract.
Missing or mistyped args	Reject	Live models invent keys.
Forbidden secret-shaped keys	Reject	Tickets are logged in plaintext JSONL.
Path outside the workspace	Reject	Side projects live in one folder.
Contract ok, execute=False	Stage	Human or test review still pending.
Contract ok, no live adapter	Block	Fail closed.
Contract ok, live adapter set	Execute	Only this row may touch the network.

The table is the product of the weekend. The Python is just a way to enforce it.

Where a free model lane belongs

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

The bus does not require a hosted model. A free model lane becomes useful only after pending() looks boring: same tools, same keys, no escaped paths. At that point a live adapter can point at a remote endpoint. MonkeyCode's free model access and free server option can host that live lane so the laptop is not both the agent and the dummy API. The bus, the ledger, and the tests stay local.

Do not put secrets in tickets. Do not treat a free lane as a production SLA. The adapter above is a weekend stub with a ten-second timeout and no retries.

What this weekend skipped

Several tempting features were written on a sticky note and then ignored.

• Replay of staged tickets into a real filesystem write. Replay needs a second review step.
• Canonical JSON hashing for cache hits. Hashing is a different article.
• Streaming model output. The bus only sees completed tool argument objects.
• Redaction of values already inside file contents. The forbidden-key check is shallow.
• A web UI for the JSONL file. cat is enough for one developer.

Skipping those items is the difference between a demo and a half-finished platform. The demo is the point.

Limitations and who should not use this

The contract table is tiny. It understands str and a couple of path rules. Nested objects, arrays, and vendor-specific tool schemas are out of scope. The ledger is append-only plaintext. Anyone with disk access can read every staged URL and path.

This approach is a poor fit for:

• Multi-tenant agents that handle other people's data.
• Workloads that need an audit log with integrity protection.
• Tools that must run with real credentials on the first call.
• Long-running workers that already have a queue and a poison-letter policy.

Those projects need a real workflow engine. A weekend JSONL file will lie to them.

The live HTTP helper also has no backoff, no idempotency key, and no TLS pinning beyond what the Python runtime provides. Treat it as a teaching adapter. Replace it before any shared environment sees traffic.

Closing

Stage the tools first. Execute later. A Sunday-night side project stays recoverable when every model-proposed call has a ticket, a contract result, and a line on disk. Wire a free remote lane only after the dry-run log is dull. The dull log is the win.