35 min

The runner

Isolation comes from the fresh context, not from a separate process.

Milestone One runner process serves three cards with a context each, and a ghost call is detected and counted.

The concierge now knows who should handle a request. Now we build what comes after that.

The obvious way would be one service per tiny agent, one container, one health check, one port. With twenty capabilities that is twenty containers, and operating them eats more time than the thing itself.

This course builds it differently, based on a single observation.

The goal

Isolation comes from the fresh context, not from a separate process. A tiny agent is not an application but a configuration: system prompt, tool selection, output schema, budget. What protects it is that its context starts empty on every invocation - and that needs no second process.

ModelRAMIsolation viaEffort
Process per agent~30 MB × Nprocesshigh: N services, N health checks
Process per request0 idleprocessmedium: ~80 ms startup
One runner, N configurations~120 MB totalcontextlow

You need process isolation once foreign code runs or two agents fight over dependencies. Neither is the case here.

Step 1: the tool registry

The runner holds all tools that exist in the system. Each card says which of them its agent may see.

# tiny/tools.py
from __future__ import annotations

import re
from typing import Callable

import httpx

DROP_TAGS = re.compile(r"<(script|style)[^>]*>.*?</\1>", re.S | re.I)
DROP_MARKUP = re.compile(r"<[^>]+>")


def http_get(url: str) -> str:
    """Fetches a URL and returns the raw text."""
    return httpx.get(url, timeout=30, follow_redirects=True).text


def html_to_text(html: str) -> str:
    """Throws away markup and leaves the readable text."""
    return re.sub(r"\n{3,}", "\n\n", DROP_MARKUP.sub(" ", DROP_TAGS.sub(" ", html))).strip()


def docs_search(query: str) -> str:
    """Placeholder for your internal search - here a file in the docs/ folder."""
    from pathlib import Path

    hits = [
        f"### {p.name}\n{p.read_text(encoding='utf-8')[:2000]}"
        for p in Path("docs").glob("*.md")
        if query.lower().split()[0] in p.read_text(encoding="utf-8").lower()
    ]
    return "\n\n".join(hits) or "No hits."


# name -> (function, JSON schema for the tools parameter)
REGISTRY: dict[str, tuple[Callable, dict]] = {
    "http_get": (
        http_get,
        {
            "type": "function",
            "function": {
                "name": "http_get",
                "description": "Fetches a URL and returns the page content.",
                "parameters": {
                    "type": "object",
                    "properties": {"url": {"type": "string"}},
                    "required": ["url"],
                },
            },
        },
    ),
    "html_to_text": (
        html_to_text,
        {
            "type": "function",
            "function": {
                "name": "html_to_text",
                "description": "Strips markup from HTML and returns the text.",
                "parameters": {
                    "type": "object",
                    "properties": {"html": {"type": "string"}},
                    "required": ["html"],
                },
            },
        },
    ),
    "docs_search": (
        docs_search,
        {
            "type": "function",
            "function": {
                "name": "docs_search",
                "description": "Searches the internal documentation for a term.",
                "parameters": {
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                    "required": ["query"],
                },
            },
        },
    ),
}


def schemas_for(names: list[str]) -> list[dict]:
    """Only these tools go into the schema. Never more, never others."""
    missing = [n for n in names if n not in REGISTRY]
    if missing:
        raise KeyError(f"Card requires unknown tools: {missing}")
    return [REGISTRY[n][1] for n in names]

That last function is the actual point of this chapter. It is three lines long and it makes the difference between a context with 200 tools and one with two.

That it raises on unknown names instead of skipping them is deliberate: a card requiring a tool that does not exist is broken. Quietly injecting fewer tools would produce an agent that works badly for no visible reason.

Step 2: the system prompt from the card

The agent does not get a hand-written prompt. It gets its own card, turned into sentences:

# tiny/runner.py
from __future__ import annotations

import json

from tiny.cards import Card
from tiny.llm import chat
from tiny.tools import REGISTRY, schemas_for


def system_prompt(card: Card) -> str:
    forbidden = "\n".join(f"- {line}" for line in card.does_not)
    fields = ", ".join(card.returns_required)
    return (
        f"You are '{card.id}', a specialised agent.\n\n"
        f"TASK\n{card.summary}\n\n"
        f"WHAT YOU DO NOT DO\n{forbidden}\n\n"
        f"OUTPUT\n"
        f"Answer at the end with a single JSON object containing exactly these "
        f"fields: {fields}. No prose before or after it.\n"
        f"If you consider yourself not responsible for the task, return "
        f'{{"status": "refused"}}.\n\n'
        f"Be brief. Your job is not to report everything but to condense what "
        f"was asked for."
    )

That last sentence is the self-description of a tiny agent, and in chapter 06 you learn why on its own it is worth nothing.

Step 3: the invocation loop

# tiny/runner.py (continued)
class Result(dict):
    """What the agent returns - plus what the run cost."""


def invoke(card: Card, task: str, counters: dict | None = None) -> Result:
    counters = counters if counters is not None else {}
    tools = schemas_for(card.tools)
    allowed = set(card.tools)

    # FRESH. No dialogue history, no state, nothing from the last invocation.
    ctx = [
        {"role": "system", "content": system_prompt(card)},
        {"role": "user", "content": task},
    ]

    for _ in range(card.budget.max_tool_calls):
        response = chat(ctx, tools=tools, max_tokens=card.budget.max_output_tokens)
        message = response["message"]
        ctx.append(message)

        calls = message.get("tool_calls") or []
        if not calls:
            return Result(
                status="ok",
                text=message.get("content", ""),
                usage=response["usage"],
            )

        for call in calls:
            name = call["function"]["name"]
            arguments = json.loads(call["function"]["arguments"] or "{}")

            if name not in allowed:
                # GHOST CALL: a tool this agent does not have.
                counters["ghost_calls"] = counters.get("ghost_calls", 0) + 1
                counters.setdefault("ghost_names", []).append(name)
                content = (
                    f"ERROR: you do not have the tool '{name}'. "
                    f"Available to you are: {sorted(allowed)}."
                )
            else:
                function = REGISTRY[name][0]
                try:
                    content = str(function(**arguments))
                except Exception as error:                      # noqa: BLE001
                    content = f"ERROR calling {name}: {error}"

            ctx.append(
                {
                    "role": "tool",
                    "tool_call_id": call["id"],
                    "content": content[:12000],   # emergency brake, the real fix follows
                }
            )

    return Result(status="partial", text="", reason="max_tool_calls reached")

Three places in there matter more than they look.

The fresh context. ctx is created inside the function and dies with it. No state, no cache, no history. That is precisely the isolation other architectures build containers for.

The ghost call counter. A model calling a tool it does not have is the single most common failure in tool calling. Most systems swallow it: they return an error message and move on. We do that too - but we count. The number is one of the few directly comparable metrics in this field, and here it falls out of normal operation for free.

The 12,000-character emergency brake. It is deliberately ugly, because it is a placeholder. It stops a single scrape from eating the tiny agent's own context - the protection applies not only to the main context. In chapter 07 it becomes something sensible: a handle instead of a truncation.

Step 4: connect concierge and runner

# tiny/runner.py (end)
def delegate(concierge, task: str, counters: dict | None = None) -> Result:
    decision = concierge.route(task)
    if decision.kind == "abstain":
        return Result(status="abstain", text="I have nobody for that.")
    card = concierge.cards[decision.agent_id]
    if card.side_effects != "none":
        # Writing agents do not run unasked. In the course the reply to the
        # caller is enough, in production the host asks the user.
        return Result(status="needs_confirmation", agent=card.id)
    result = invoke(card, task, counters)
    result["agent"] = card.id
    result["routing_score"] = round(decision.score, 4)
    return result

The side_effects check is four lines and prevents the nastiest class of failure in such systems: a router that is wrong writing something while it is wrong. A misrouted read costs tokens. A misrouted write costs a record.

The milestone

# scripts/05_runner_test.py
from tiny.concierge import load_concierge
from tiny.runner import delegate

concierge = load_concierge()
counters: dict = {}

for task in [
    "read https://en.wikipedia.org/wiki/Tool and tell me in two sentences what it is about",
    "how did the onboarding process work again",
    "please add mustermann gmbh as a new customer",
    "make me a sandwich",
]:
    r = delegate(concierge, task, counters)
    print(f"[{r['status']:<18}] {r.get('agent', '-'):<14} {task[:52]}")

print(f"\nGhost calls: {counters.get('ghost_calls', 0)} {counters.get('ghost_names', [])}")
[ok                ] web-reader     read https://en.wikipedia.org/wiki/Tool and tell me ...
[ok                ] docs-finder    how did the onboarding process work again
[needs_confirmation] crm-writer     please add mustermann gmbh as a new customer
[abstain           ] -              make me a sandwich

Ghost calls: 1 ['web_search']

The state is reached when three things hold at once: one runner process serves all three cards, every invocation sees only its own one or two tools, and the ghost call counter shows not an arbitrary value but what actually happened.

If your counter is high, that is not a failure. Small models like to invent web_search because they know it from training. Note the number down - in chapter 08 you compare it against the variant where all tools sit in one context. That is one of the points where the whole thesis is decided.