DEV Community

Morgan Sun
Morgan Sun

Posted on

Put a Capability Broker Between Your Agent and Its Tools

#ai #agents #security #python

Agent incidents rarely look like a movie villain. They look like a helpful assistant combining two permissions you never meant to combine: a note-reader that can summarize a folder, plus a messenger that can post to a webhook, plus one poisoned paragraph in a document that says where the notes should go.

After writing about prompt changes as migrations, I started treating tool permission the same way: as a reviewed contract, not a vibe in the system prompt. The pattern I keep reaching for is a capability broker. The model is allowed to ask for an action. It is not allowed to perform one.

Design rule: intent leaves the model, authority stays in code

The flow is deliberately plain:

  1. The agent returns a structured intent such as {'tool': 'fs.read', 'args': {'path': 'notes/trip.md'}}.
  2. The broker resolves paths, hosts, recipients, and flags before anything touches the OS or network.
  3. A manifest decides allow, deny, or needs-approval.
  4. The model receives either sanitized data or a normal-looking tool error.
  5. Every decision is appended to a local audit table.

This changes the security question from "will the model stay polite?" to "did the policy admit this exact call?" That second question is testable.

# capabilities.toml
schema = 1

[tools.fs_read]
roots = ['./workspace', './public_data']
never = ['*.pem', '.env', './private']

[tools.http_fetch]
hosts = ['api.github.com', 'raw.githubusercontent.com']
max_bytes = 200000

[tools.notify]
webhooks = ['https://hooks.internal.example/*']
approval = 'human'
Enter fullscreen mode Exit fullscreen mode 
# broker.py
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
import fnmatch, json, sqlite3, tomllib

class Verdict(Enum):
    ALLOW = 'allow'
    DENY = 'deny'
    APPROVAL = 'approval'

@dataclass
class Decision:
    verdict: Verdict
    reason: str = ''

class Broker:
    def __init__(self, manifest_path='capabilities.toml', db='audit.sqlite3'):
        self.cfg = tomllib.loads(Path(manifest_path).read_text())
        self.db = sqlite3.connect(db)
        self.db.execute('create table if not exists audit(tool text, args text, verdict text, reason text)')

    def _record(self, tool, args, decision):
        self.db.execute('insert into audit values (?,?,?,?)', (tool, json.dumps(args), decision.verdict.value, decision.reason))
        self.db.commit()

    def _path_ok(self, raw):
        p = Path(raw).expanduser().resolve()
        root_hits = [Path(r).expanduser().resolve() for r in self.cfg['tools']['fs_read']['roots']]
        if not any(p == r or r in p.parents for r in root_hits):
            return Decision(Verdict.DENY, 'outside approved roots')
        s = str(p)
        if any(fnmatch.fnmatch(s, pat) or p.name == pat for pat in self.cfg['tools']['fs_read']['never']):
            return Decision(Verdict.DENY, 'matches never list')
        return Decision(Verdict.ALLOW)

    def review(self, tool, args):
        table = self.cfg['tools'].get(tool)
        if table is None:
            d = Decision(Verdict.DENY, 'tool is not declared')
        elif tool == 'fs_read':
            d = self._path_ok(args.get('path', ''))
        elif tool == 'http_fetch':
            host = args.get('host', '')
            d = Decision(Verdict.ALLOW) if host in table['hosts'] else Decision(Verdict.DENY, 'host not declared')
        elif tool == 'notify':
            url = args.get('url', '')
            ok = any(fnmatch.fnmatch(url, pat) for pat in table['webhooks'])
            d = Decision(Verdict.APPROVAL, 'human signoff required') if ok and table.get('approval') == 'human' else Decision(Verdict.DENY, 'webhook not declared')
        else:
            d = Decision(Verdict.DENY, 'no reviewer implemented')
        self._record(tool, args, d)
        return d

    def invoke(self, tool, args):
        d = self.review(tool, args)
        if d.verdict is not Verdict.ALLOW:
            return {'ok': False, 'error': 'tool unavailable'}  # do not leak policy detail
        return run_real_tool(tool, args)  # inject implementations here
Enter fullscreen mode Exit fullscreen mode

The important behavior is that refusal is boring. If the agent asks for ./workspace/../../.env, it gets the same bland tool unavailable shape as a transient failure. It can choose another action, but it cannot learn which rule fired or negotiate with the enforcer.

Tests should attack the broker, not praise the model

A small pytest matrix is more useful than a clever prompt:

import pytest
from broker import Broker, Verdict

@pytest.mark.parametrize('path,verdict', [
    ('workspace/report.md', Verdict.ALLOW),
    ('workspace/../.env', Verdict.DENY),
    ('public_data/../private/payroll.csv', Verdict.DENY),
    ('workspace/key.pem', Verdict.DENY),
])
def test_fs_boundaries(tmp_path, path, verdict):
    b = Broker()
    assert b.review('fs_read', {'path': path}).verdict is verdict

def test_undeclared_tool_is_closed():
    assert Broker().review('shell.exec', {'cmd': 'ls'}).verdict is Verdict.DENY

def test_approval_is_not_silent_allow():
    b = Broker()
    d = b.review('notify', {'url': 'https://hooks.internal.example/alerts'})
    assert d.verdict is Verdict.APPROVAL
Enter fullscreen mode Exit fullscreen mode

Then add an adversarial corpus: documents that beg, impersonate users, claim the policy changed, or encode instructions in markdown comments. The assertion is never "the model refused." The assertion is that if the model emits a forbidden intent, the audit table contains a denial and no real side effect happened. A well-behaved model is nice; a broker that still says no is the control.

A cheap place to practice the loop

You do not need a large budget to red-team this shape. You need an endpoint you can hit repeatedly with odd inputs while the manifest evolves. Disclosure: This article was prepared as part of MonkeyCode's product outreach. I have used MonkeyCode's free model access and free server option as a low-friction place to run these adversarial loops during development; treat that as an availability note, not a benchmark, quota promise, or production recommendation, and check their current docs before building around it.

If the fit is right, use it for rehearsal: seed malicious documents, regenerate intents, diff the audit output, tighten the manifest. For anything customer-facing, keep the same broker and swap the model behind it based on latency, data handling, and compliance needs. The broker should not care which model produced the intent, because intent is untrusted either way.

Where this breaks

  • It is not a sandbox. A safe allowlist cannot rescue a vulnerable fs_read implementation, a leaked token inside a tool, or a parser bug. Put serious workloads behind containers, users, seccomp/AppArmor, and network egress rules too.
  • Path checks are easy to under-specify. Resolve before matching, think about symlinks and case-insensitive filesystems, and fuzz separators, Unicode, .., and trailing slashes.
  • Approval fatigue is real. If every notify action needs a human click, people will start auto-approving. Reserve approval for irreversible or externally visible actions.
  • Fixed pipelines gain little. If your workflow never lets the model choose tools freely, static routing may already give you the same guarantee with less machinery.
  • Regulated deletion, payments, production deploys, and medical/legal actions need more than globs: dual control, rate limits, tamper-evident logs, and sometimes a formal review.

The durable idea is simple: keep judgment-heavy text generation away from authority-heavy execution. Let the model propose. Let boring code dispose. Version the policy, replay the audit, and make denial indistinguishable from an ordinary bad day for a tool.

Top comments (0)