toolgate-ai 1.1.1

Optimistic execution middleware for autonomous agents — let reads pass, intercept actions, approve at the end. Supports MCP servers and custom local/hosted tools.

ToolGate (Python)

Execution middleware for autonomous agents — intercept, approve, execute.

Every tool call your agent makes goes through ToolGate. Reads pass through instantly. Writes, sends, deletes, and executes are intercepted, return a phantom response so the agent keeps running, and queue for human approval. Approved actions execute for real — even after the agent process has stopped.

Agent calls tool ──▶ ToolGate classifies ──▶ READ?   ──▶ Execute immediately, record result
                                           │
                                           └──▶ ACTION? ──▶ Return phantom, store params
                                                              ┊
                                           Agent finishes ──▶ Session in dashboard
                                                              ┊
                                           From anywhere  ──▶ Approve / Edit / Reject
                                                              (executes for real)

Works with MCP servers, custom local tool functions, and hosted tool endpoints.

Also available as an npm package for TypeScript/JavaScript.

Install

pip install toolgate-ai

Two ways to use ToolGate

1 — MCP servers

Wrap any MCP server. ToolGate auto-discovers tools, classifies them, and intercepts actions.

import asyncio
from toolgate import mcp_tool_gate, auto_config_from_mcp

async def main():
    config = await auto_config_from_mcp([
        {"name": "gmail",  "url": "https://gmail-mcp.example.com",  "headers": {"Authorization": "Bearer ya29.xxx"}},
        {"name": "github", "url": "https://github-mcp.example.com", "headers": {"X-API-Key": "ghp_xxx"}},
    ])

    gate = mcp_tool_gate(
        mcp_executor,
        api_key="tg_live_...",
        agent_name="inbox-agent",
        mcp_servers=config.mcp_servers,
    )

    gate.describe("Read inbox and draft replies to urgent emails")

    emails = await gate.proxy("gmail_listMessages", {"maxResults": 10})  # ✅ READ — runs immediately
    draft  = await gate.proxy("gmail_sendMessage", {"to": "...", "body": "..."})  # 🛑 SEND — intercepted

    await gate.finalize()
    # Session appears in dashboard. Click Approve & Execute to send the real email.

asyncio.run(main())

2 — Custom tools (decorator style)

No MCP server needed. Use the @gate.tool decorator on any async function.

from toolgate import ToolGate

gate = ToolGate(executor, api_key="tg_live_...")

@gate.tool(kind="read")          # force passthrough
async def read_file(params):
    """Read a file from disk"""
    return open(params["path"]).read()

@gate.tool                        # auto-classified as ACTION (write*)
async def write_file(params):
    """Write content to a file"""
    open(params["path"], "w").write(params["content"])

@gate.tool
async def delete_file(params):
    """Delete a file"""
    import os; os.unlink(params["path"])

# Use the decorated functions — ToolGate intercepts where needed
content = await read_file({"path": "config.json"})           # ✅ executes immediately
await write_file({"path": "out.txt", "content": "hello"})    # 🛑 intercepted → approval

await gate.finalize()
await gate.approve_all()  # or let the dashboard handle it

Custom tools (wrap_tools style)

Wrap a dict of existing functions without changing their definitions:

from toolgate import ToolGate

gate = ToolGate(executor, api_key="tg_live_...")

tools = gate.wrap_tools(
    {
        "read_file":   read_file_fn,
        "write_file":  write_file_fn,
        "delete_file": delete_file_fn,
    },
    overrides={
        "read_file": {"kind": "read", "description": "Read a file from disk"},
    },
)

# Give `tools` to your agent
content = await tools["read_file"]({"path": "config.json"})   # ✅ passthrough
await tools["write_file"]({"path": "out.txt", "content": "hello"})  # 🛑 intercepted

Custom tools (ToolDef config style)

from toolgate import ToolGate, ToolDef

gate = ToolGate(
    executor,
    api_key="tg_live_...",
    tools=[
        ToolDef(name="read_file",   description="Read a file",   kind="read", fn=read_file_fn),
        ToolDef(name="write_file",  description="Write a file",               fn=write_file_fn),
        ToolDef(name="delete_file", description="Delete a file",              fn=delete_file_fn),
    ],
)

Mixing MCP + custom tools

gate = ToolGate(
    mcp_executor,
    api_key="tg_live_...",
    mcp_servers=[MCPServerDef(name="gmail", url="...", tools=[...])],
    tools=[ToolDef(name="read_file", description="Read a local file", kind="read", fn=read_file_fn)],
)

---

Local vs Hosted

	Local (free)	Hosted (paid)
Custom tools	Provide fn — called in-process after approval	Provide endpoint — dashboard POSTs to your URL after approval
MCP servers	Local / stdio MCP	ToolGate hosted MCP proxy
Agent must be running for execution?	Yes	No — deferred execution, agent can be offline

Hosted custom tool endpoint

from toolgate import ToolGate, ToolDef

gate = ToolGate(
    executor,
    api_key="tg_live_...",
    tools=[
        ToolDef(
            name="send_report",
            description="Email a weekly summary report",
            endpoint="https://myapp.com/hooks/send_report",
            headers={"Authorization": "Bearer secret"},
        ),
    ],
)

When the human approves in the dashboard, ToolGate POSTs to your endpoint:

POST https://myapp.com/hooks/send_report
Content-Type: application/json
Authorization: Bearer secret

{ "tool": "send_report", "params": { ... } }

No agent process needs to be running. The result is saved back to the session.

---

Classification

ToolGate automatically classifies every tool call as a read (passthrough) or action (intercept) by analyzing the tool name and description:

• list*, get*, search*, fetch*, read* → read (passthrough)
• send*, create*, delete*, update*, write*, post* → action (intercepted)
• Unknown → intercepted (safe default)

Override classification:

# In ToolDef
ToolDef(name="my_tool", description="...", kind="read")    # always passthrough
ToolDef(name="my_tool", description="...", kind="action")  # always intercepted

# In constructor
gate = ToolGate(
    executor,
    read_tools=["always_pass"],
    action_tools=["always_intercept"],
    classifier=lambda name, params: ToolClassification(ToolIntent.WRITE, False, 1.0, "custom"),
)

---

Getting Your API Key

1. Sign up at toolgate.dev
2. Go to API Keys → + New Key
3. Copy the key — no database setup required

---

Programmatic Approval

await gate.finalize()

# Approve everything
await gate.approve_all()

# Or reject everything
await gate.reject_all()

# Or per-action
from toolgate import ApprovalResult, ActionDecision

await gate.execute_approval(ApprovalResult(
    session_id=gate.session_id,
    approved_at=time.time() * 1000,
    decisions={
        "action-id-1": ActionDecision(action="approve"),
        "action-id-2": ActionDecision(action="reject"),
        "action-id-3": ActionDecision(action="edit", new_params={"to": "[e-mail]"}),
    },
))

---

Inspection

gate.reads            # all reads that executed (with results)
gate.pending          # all intercepted actions
gate.current_session  # full session snapshot
gate.summary()        # pretty-printed CLI summary — shows [local-tool] / [hosted-tool] / [local-mcp] / [hosted-mcp]

---

API Reference

ToolGate(executor, *, api_key, agent_name, tools, mcp_servers, ...)

Param	Type	Description
api_key	str	ToolGate API key — required for dashboard sync
agent_name	str	Display name shown in dashboard
tools	list[ToolDef]	Custom tool definitions (local or hosted)
mcp_servers	list[MCPServerDef]	MCP server definitions
read_tools	list[str]	Tool names that always passthrough
action_tools	list[str]	Tool names that always intercept
classifier	(name, params) -> ToolClassification	Custom classifier override
phantom_response	(ToolCall) -> Any	Custom phantom response factory

ToolDef

Field	Type	Description
name	str	Must match what your agent calls
description	str	Used by the classifier heuristic
kind	"read" | "action"	Force classification (omit to auto-classify)
fn	async (params) -> Any	Local: called in-process after approval (free)
endpoint	str	Hosted: dashboard POSTs here after approval (paid)
headers	dict[str, str]	Auth headers for hosted endpoint

MCPServerDef

Field	Type	Description
name	str	Display name
url	str	MCP server HTTP endpoint
tools	list[MCPToolDef]	Tool list (omit to use auto_config_from_mcp)
headers	dict[str, str]	Auth headers sent on every call

@gate.tool decorator

@gate.tool                             # auto-classify from name/description
@gate.tool(kind="read")               # force read (passthrough)
@gate.tool(kind="action")             # force action (intercept)
@gate.tool(description="My tool")     # custom description for classifier
@gate.tool(endpoint="https://...")    # hosted tool — dashboard calls this URL
@gate.tool(endpoint="...", headers={}) # hosted tool with auth

gate.wrap_tools(tools, overrides={})

Wraps a { name: fn } dict. Returns proxied callables. Accepts overrides per tool name with kind and description.

mcp_tool_gate(executor, *, api_key, mcp_servers, agent_name)

Convenience factory — equivalent to ToolGate(executor, mcp_servers=...). Useful when working exclusively with MCP.

auto_config_from_mcp(servers)

Calls tools/list on each server and returns a ToolGateConfig. Accepts headers per server.

Instance methods

Method	Description
.proxy(name, params)	Executor to hand to your agent
.wrap_tools(fns, overrides)	Wrap a {name: fn} dict — returns proxied callables
.describe(text)	Set a task description shown in the dashboard
.finalize()	End session, persist to dashboard
.approve_all()	Approve + execute all pending actions locally
.reject_all()	Reject all pending actions
.execute_approval(result)	Execute with per-action decisions
.summary()	Pretty-printed session summary

---

License

MIT