Robust tools & structured I/O

schema.py# The tool SCHEMA is the prompt the model reads to decide when/how to call it.
# A vague name and a fuzzy description make the model guess. Here we render
# the same tool two ways and show what the model effectively "sees".

bad = {
    'name': 'do_lookup',
    'description': 'looks stuff up',
    'parameters': {'q': 'string', 'opt': 'string'},   # untyped, unexplained
}

good = {
    'name': 'get_order_status',
    'description': ('Return the current status of ONE order by its id. '
                    'Use when the user asks where an order is or if it shipped. '
                    "Example: get_order_status(order_id='A-1001')."),
    'parameters': {
        'order_id': {'type': 'string', 'required': True,
                     'description': "Order id like 'A-1001'."},
    },
}

def render(schema):
    lines = [f"{schema['name']}: {schema['description']}"]
    for pname, pinfo in schema['parameters'].items():
        if isinstance(pinfo, dict):
            req = 'required' if pinfo.get('required') else 'optional'
            lines.append(f"  - {pname} ({pinfo['type']}, {req}): {pinfo['description']}")
        else:
            lines.append(f'  - {pname} ({pinfo}, ?): ?')
    return '\n'.join(lines)

print('WHAT THE MODEL SEES (bad):')
print(render(bad))
print('\nWHAT THE MODEL SEES (good):')
print(render(good))

validate.py# A validate-and-coerce wrapper: NEVER trust the model's tool args.
# Real code uses pydantic / JSON-schema; this is a stdlib stand-in that
# checks types, ranges, and required fields, coercing where it safely can.

SCHEMA = {
    'city':  {'type': str, 'required': True},
    'days':  {'type': int, 'required': True, 'min': 1, 'max': 14},
    'units': {'type': str, 'required': False, 'default': 'celsius',
              'choices': ['celsius', 'fahrenheit']},
}

class ToolArgError(Exception):
    pass

def validate(args, schema):
    clean = {}
    for name, rule in schema.items():
        if name not in args:
            if rule.get('required'):
                raise ToolArgError(f'missing required arg: {name!r}')
            clean[name] = rule.get('default')
            continue
        val = args[name]
        want = rule['type']
        # coerce a stringified number the model may have emitted ('3' -> 3)
        if want is int and isinstance(val, str):
            try:
                val = int(val)
            except ValueError:
                raise ToolArgError(f'{name!r} must be an int, got {val!r}')
        if not isinstance(val, want):
            raise ToolArgError(f'{name!r} must be {want.__name__}, got {type(val).__name__}')
        if 'min' in rule and val < rule['min']:
            raise ToolArgError(f'{name!r}={val} below min {rule["min"]}')
        if 'max' in rule and val > rule['max']:
            raise ToolArgError(f'{name!r}={val} above max {rule["max"]}')
        if 'choices' in rule and val not in rule['choices']:
            raise ToolArgError(f'{name!r}={val!r} not in {rule["choices"]}')
        clean[name] = val
    return clean

# The model emitted these tool args. Note: days is the STRING '3' (it happens),
# and there is no 'units' key at all.
model_args = {'city': 'Berlin', 'days': '3'}
print('clean:', validate(model_args, SCHEMA))

# Now a hallucinated / out-of-range arg the model made up:
bad = {'city': 'Berlin', 'days': 99}
try:
    validate(bad, SCHEMA)
except ToolArgError as e:
    print('rejected:', e)

structured.py# Getting reliable JSON out of a model. Real APIs offer schema-constrained /
# 'structured output' modes; here we SIMULATE the model's reply as a fixed
# string, then parse + validate it and RE-ASK once if it doesn't conform.
import json

EXPECTED = {'sentiment': str, 'score': float, 'topics': list}

def validate_json(text, expected):
    """Parse text as JSON and check it matches the expected shape.
    Returns (ok, value_or_error)."""
    try:
        obj = json.loads(text)
    except json.JSONDecodeError as e:
        return False, f'not valid JSON: {e.msg}'
    if not isinstance(obj, dict):
        return False, 'top level is not an object'
    for key, typ in expected.items():
        if key not in obj:
            return False, f'missing key: {key!r}'
        if not isinstance(obj[key], typ):
            return False, f'{key!r} should be {typ.__name__}'
    return True, obj

# Two SIMULATED model replies. The first is what a model often does on the
# first try: wraps JSON in prose / a code fence. The second is clean (the
# reply after we re-ask with a stricter instruction).
reply_1 = 'Sure! Here is the analysis:\n```json\n{"sentiment": "negative"}\n```'
reply_2 = '{"sentiment": "negative", "score": 0.82, "topics": ["latency", "billing"]}'

def ask_model(attempt):
    return reply_1 if attempt == 0 else reply_2

result = None
for attempt in range(2):
    ok, val = validate_json(ask_model(attempt), EXPECTED)
    print(f'attempt {attempt}: ok={ok} -> {val}')
    if ok:
        result = val
        break
    # repair step: on failure we re-ask the model with the error appended
    # so it can correct itself. Here attempt 1 returns clean JSON.

print('final:', result)

deps.py# A tiny dependency resolver. Some tool calls are INDEPENDENT (can run
# 'together'); some DEPEND on another's output and must be sequenced.
# No real async here — we simulate execution and show the parallel GROUPS.

# Each tool call: an id, the tool, and the ids it depends on.
calls = [
    {'id': 'A', 'tool': 'get_user',    'deps': []},
    {'id': 'B', 'tool': 'get_weather', 'deps': []},          # independent of A
    {'id': 'C', 'tool': 'get_orders',  'deps': ['A']},       # needs the user
    {'id': 'D', 'tool': 'summarize',   'deps': ['B', 'C']},  # needs both
]

def plan_groups(calls):
    """Return a list of GROUPS; every call in a group can run in parallel
    because all its deps were satisfied by an earlier group (Kahn-style)."""
    by_id = {c['id']: c for c in calls}
    done, groups = set(), []
    remaining = list(by_id)
    while remaining:
        ready = [cid for cid in remaining
                 if all(d in done for d in by_id[cid]['deps'])]
        if not ready:
            raise ValueError('cycle or missing dependency')
        ready.sort()                 # deterministic order within a group
        groups.append(ready)
        done.update(ready)
        remaining = [c for c in remaining if c not in done]
    return groups

groups = plan_groups(calls)
for i, g in enumerate(groups, 1):
    kind = 'parallel' if len(g) > 1 else 'single'
    print(f'group {i} ({kind}): {", ".join(g)}')

results.py# Handling tool RESULTS before they go back to the model:
#  (1) truncate/summarize large output,
#  (2) return errors as STRUCTURED results the model can reason about,
#  (3) idempotency so a retried tool doesn't act twice.

MAX_CHARS = 80

def wrap_result(tool, ok, data):
    """Normalize every tool result into the same small dict the model reads."""
    if not ok:
        return {'tool': tool, 'status': 'error', 'error': data}
    text = str(data)
    if len(text) > MAX_CHARS:
        text = text[:MAX_CHARS] + f'... [truncated, {len(str(data))} chars total]'
    return {'tool': tool, 'status': 'ok', 'result': text}

# A large successful result gets truncated before it wastes the context budget:
big = 'customer record ' * 200
print(wrap_result('db_query', True, big))

# A failure becomes a structured error the model can act on (retry / apologize):
print(wrap_result('charge_card', False, 'gateway timeout after 30s'))

# Idempotency: a retried tool with the same key must NOT run its effect twice.
_seen = {}
def run_once(key, effect):
    if key in _seen:
        return {'status': 'cached', 'result': _seen[key]}
    _seen[key] = effect()
    return {'status': 'ran', 'result': _seen[key]}

charge = lambda: 'charged $50'
print(run_once('order-42', charge))   # first call runs the effect
print(run_once('order-42', charge))   # retry is served from cache, no double charge

sandbox.py# Tool-execution safety. A tool can delete files, spend money, or hit prod.
# Give each tool LEAST PRIVILEGE, gate destructive actions behind a confirm /
# dry-run, and allow-list what it may touch. (Timeouts noted in prose.)

# Declarative policy: what each tool is allowed to do.
POLICY = {
    'read_file':   {'destructive': False, 'paths': ['/data/']},
    'delete_file': {'destructive': True,  'paths': ['/tmp/']},
    'send_email':  {'destructive': True,  'paths': []},
}

class Denied(Exception):
    pass

def guard(tool, target=None, *, confirm=False, dry_run=True):
    rule = POLICY.get(tool)
    if rule is None:
        raise Denied(f'tool {tool!r} is not on the allow-list')
    # path allow-list: the target must sit under an allowed prefix
    if rule['paths'] and target is not None:
        if not any(target.startswith(p) for p in rule['paths']):
            raise Denied(f"{tool!r} may not touch {target!r} (allowed: {rule['paths']})")
    # destructive actions must be confirmed AND not left in dry-run
    if rule['destructive'] and not (confirm and not dry_run):
        return {'tool': tool, 'target': target, 'action': 'SKIPPED (needs confirm, not dry-run)'}
    return {'tool': tool, 'target': target, 'action': 'EXECUTED'}

# 1. safe read under an allowed path:
print(guard('read_file', '/data/report.csv'))
# 2. destructive delete but still a dry-run -> not executed:
print(guard('delete_file', '/tmp/scratch.log', confirm=True, dry_run=True))
# 3. same delete, confirmed and dry-run off -> executes:
print(guard('delete_file', '/tmp/scratch.log', confirm=True, dry_run=False))
# 4. path outside the allow-list -> denied:
try:
    guard('delete_file', '/etc/passwd', confirm=True, dry_run=False)
except Denied as e:
    print('DENIED:', e)
# 5. a tool nobody granted -> denied:
try:
    guard('run_shell', 'rm -rf /')
except Denied as e:
    print('DENIED:', e)