Mediation Layer

A runtime authorization layer that sits between an AI agent and the tools it's allowed to use. Every proposed tool call is intercepted and evaluated — ALLOW, DENY, or ESCALATE — before it ever reaches the real tool. The agent proposes an action; it can never authorize itself.

AI Agent → Mediation Layer → Tool
             │
             ├── Directive & parameter provenance (who/what told the
             │   agent to do this — user, prior tool output, or a
             │   document/webpage it read?)
             ├── Declared scope + session context
             ├── Cumulative session risk
             └── Chained-privilege-escalation detection: individually
                 in-scope steps whose COMBINED effect was never
                 authorized (e.g. read-sensitive-data → send-externally)

Distributed as mediation-layer; the Python package you import is toolgate (that's the project's internal/working name — see Mediation_Layer.md for the full design rationale behind the rule set).

Install

pip install mediation-layer

(Pre-release / from source: pip install . from the repo root, or pip install dist/mediation_layer-*.whl after building.)

from toolgate import TaskSession, guarded_tool, untrusted, ToolDenied, ToolEscalated

Quickstart

Protect your own tool functions with @guarded_tool. The function body is whatever you already had — toolgate only decides whether it runs:

from toolgate import TaskSession, guarded_tool, untrusted, ToolDenied

@guarded_tool(category="file_ops", operation="read")
def read_file(path: str) -> str:
    return open(path).read()

@guarded_tool(category="messaging", operation="send")
def send_message(to: str, body: str) -> dict:
    ...  # your real send logic
    return {"sent": True, "to": to}

with TaskSession(intent="Read the invoice and summarize it",
                 scope={"read:file_ops"}) as session:

    content = read_file(path="inbox/invoice.txt")   # in scope -> ALLOW

    # A value that came from somewhere untrusted is labelled, so the
    # policy engine knows its real origin instead of trusting the agent:
    try:
        send_message(to=untrusted("evil@example.com", origin="invoice.txt"),
                     body="...")
    except ToolDenied as e:
        print(e.result.rule, e.result.reason)

Already have plain functions (or LangChain tools) you don't want to rewrite as decorators? Wrap them instead — same enforcement, no changes to the functions themselves:

from toolgate import guard_tools

def delete_file(path: str) -> bool:
    ...  # your real delete logic, completely unaware of toolgate

# overrides is only needed when the function name doesn't hint its
# category/operation clearly enough for auto-inference to guess right
protected = {t.__name__: t for t in guard_tools(
    [delete_file],
    overrides={"delete_file": ("file_ops", "delete")},
)}

# hand `protected` to your agent's tool registry — never the raw function.
# See examples/external_agent_demo.py for this exact pattern end to end,
# including proof the agent only ever receives the wrapped version.

from toolgate.integrations.langchain import guard_langchain_tools

safe_tools = guard_langchain_tools(my_tools)      # list[BaseTool] in, list[BaseTool] out
agent = create_react_agent(llm, safe_tools)

The three decisions

Resolving an escalation:

try:
    delete_file(path="reports/old_draft.txt")
except ToolEscalated as e:
    print(e.result.reason)
    result = session.confirm(e)   # records the human override, runs the
                                   # original call exactly once — never
                                   # re-argued by the content that
                                   # triggered the escalation

session.confirm(e, reusable=True) lets an identical later call skip re-asking; anything different (destination, resource, phase, scope, directive source, or risk state) is evaluated fresh, never covered by an old approval.

Chained privilege escalation

Each step below is individually within the declared scope. The mediator still catches the combined effect:

with TaskSession(intent="Read the summary and notify the auditor",
                 scope={"read:file_ops", "query:db_access", "send:messaging"}):
    read_file(path="reports/q3_summary.txt")              # ALLOW
    query_db(table="employees", columns="name,salary")    # ALLOW (tags read_sensitive)
    send_message(to="auditor@vendor-x.com", body=...)      # ESCALATE:
                                                            # "data-exfiltration" chain

See examples/chained_escalation.py for the full scenario, including the same chain driven by an injected document (denied outright) and the same chain pre-authorized by an explicit scope tag (allowed straight through).

Dashboard

import toolgate
toolgate.serve_dashboard()            # background thread, http://127.0.0.1:8000

Reads the same in-process store every TaskSession writes to — decisions from @guarded_tool, guard_tools(), and framework adapters all show up with no extra wiring. GET /audit/recent is the aggregated feed the dashboard itself polls, if you want the raw JSON.

Examples

• examples/quickstart.py — the pattern above, runnable
• examples/chained_escalation.py — the chain scenario, all four variants
• examples/stale_authorization.py — why a human approval doesn't become a standing permission
• examples/external_agent_demo.py — a plain, non-framework tool-calling agent using only guard_tools(), proving the agent never receives the unwrapped tool
• examples/langchain_agent.py + examples/agent_server.py — a real LangChain agent behind a chat UI with the dashboard embedded

Optional extras

pip install "mediation-layer[llm]"        # live LLM advisor (Groq or Anthropic) for ambiguous-scope
                                           # and injection checks; falls back to a deterministic
                                           # heuristic with no API key set, so it always runs offline
pip install "mediation-layer[langchain]"  # guard_langchain_tools
pip install "mediation-layer[dev]"        # pytest, httpx (for the test suite)

The LLM advisor is consulted only for two narrow, confidence-scored questions (ambiguous scope match, suspected injected instruction) and never returns an authorization decision itself — the rule-based policy engine is always the final word.

Testing

pip install -e ".[dev,langchain]"
pytest tests/ -v

License

MIT