Optimistic execution middleware for autonomous agents — let reads pass, intercept actions, approve at the end.
Project description
🛡️ ToolGate (Python)
Optimistic execution middleware for autonomous agents.
Let your agent run freely — reads execute instantly, every other action (writes, deletes, sends, deploys) is silently intercepted, recorded, and held for human approval. The agent never knows the difference.
Agent calls tool ──▶ ToolGate classifies ──▶ READ? ──▶ Execute for real
│
└──▶ ACTION? ──▶ Return phantom result
Save params to queue
┊
Agent finishes ──▶ Human reviews all actions
┊
Approve ──▶ Execute for real
Reject ──▶ Discard
Edit ──▶ Execute with new params
Also available as an npm package for TypeScript/JavaScript.
Install
pip install toolgate-ai
Quick Start
import asyncio
from toolgate import ToolGate, cli_approval
# Your real tool executor (what the agent normally calls)
async def execute(tool: str, params: dict) -> any:
# call your APIs, MCP servers, SDKs, whatever
...
async def main():
gate = ToolGate(
execute,
api_key="tg_live_your_api_key_here", # get one from the dashboard
agent_name="my-agent",
on_approval_needed=cli_approval, # interactive terminal approval
)
# Give your agent gate.proxy instead of the real executor
result = await gate.proxy("listEmails", {})
# When done, finalize — triggers the approval flow
await gate.finalize()
asyncio.run(main())
That's it. Your agent runs as normal. Reads pass through. Actions are queued. You approve at the end. Sessions automatically appear in your dashboard.
Getting Your API Key
- Sign up at the ToolGate Dashboard
- Go to API Keys and click + New Key
- Copy the key (shown only once) and use it in your code:
gate = ToolGate(execute, api_key="tg_live_a1b2c3d4e5f6...")
That's all you need — no database credentials, no backend setup.
How Classification Works
ToolGate uses a multi-layer classifier to determine if a tool call is a read (passthrough) or an action (intercept):
- Explicit overrides —
read_tools=["getUser"]/action_tools=["deleteUser"] - Custom classifier — provide your own function
- Name pattern matching —
get*,list*,search*→ read.create*,send*,delete*→ action - MCP description analysis — parses tool descriptions for read vs. mutate signals
- HTTP method hints — if params contain
method: "GET"→ read,method: "POST"→ action - Safe default — unknown tools are intercepted (never accidentally executes a write)
Explicit Tool Lists
gate = ToolGate(
execute,
api_key="tg_live_...",
read_tools=["getEmails", "searchContacts", "listFiles"],
action_tools=["sendEmail", "deleteFile", "createIssue"],
)
Custom Classifier
from toolgate import ToolGate, classify_tool, ToolClassification, ToolIntent
def my_classifier(tool_name: str, params: dict) -> ToolClassification:
if tool_name.startswith("db.query"):
return ToolClassification(
intent=ToolIntent.READ,
is_passthrough=True,
confidence=1.0,
reason="DB query",
)
return classify_tool(tool_name, params)
gate = ToolGate(execute, api_key="tg_live_...", classifier=my_classifier)
MCP Integration
from toolgate import mcp_tool_gate, auto_config_from_mcp, MCPServerDef, MCPToolDef
# Option 1: Provide tool definitions upfront
gate = mcp_tool_gate(
mcp_executor,
api_key="tg_live_...",
mcp_servers=[
MCPServerDef(
name="gmail",
url="https://gmail.mcp.example.com/sse",
tools=[
MCPToolDef(name="gmail_listMessages", description="List messages in the inbox"),
MCPToolDef(name="gmail_sendMessage", description="Send a new email message"),
],
),
],
)
# Option 2: Auto-discover from running MCP servers
config = await auto_config_from_mcp([
{"name": "gmail", "url": "https://gmail.mcp.example.com/sse"},
{"name": "github", "url": "https://github.mcp.example.com/sse"},
])
gate = ToolGate(mcp_executor, api_key="tg_live_...", config=config)
Wrapping an Existing Tool Map
real_tools = {
"getUser": get_user,
"createUser": create_user,
"sendEmail": send_email,
}
gate = ToolGate(
lambda name, params: real_tools[name](params),
api_key="tg_live_...",
agent_name="user-manager",
)
# Wrap the tool map
proxied_tools = gate.wrap_tools(real_tools)
Phantom Responses
When an action is intercepted, the agent receives a convincing phantom response:
| Intent | Default Phantom Response |
|---|---|
create |
{ "success": True, "id": "phantom_abc123" } |
update |
{ "success": True, "message": "Updated successfully" } |
delete |
{ "success": True, "message": "Deleted successfully" } |
send |
{ "success": True, "messageId": "msg_xyz789" } |
Custom phantom responses:
def my_phantom(tool):
if tool.name == "createIssue":
return {"id": 99999, "url": "https://github.com/org/repo/issues/99999"}
return {"ok": True}
gate = ToolGate(execute, api_key="tg_live_...", phantom_response=my_phantom)
Approval Methods
1. CLI (built-in)
from toolgate import ToolGate, cli_approval
gate = ToolGate(execute, api_key="tg_live_...", on_approval_needed=cli_approval)
2. Programmatic
gate = ToolGate(execute, api_key="tg_live_...")
await agent.run(tools=gate.proxy)
request = await gate.finalize()
# Inspect what the agent wants to do
print(gate.summary())
# Approve everything
await gate.approve_all()
# Or reject everything
await gate.reject_all()
3. Dashboard (web UI)
from toolgate import ToolGate, dashboard_approval, DashboardApprovalConfig
gate = ToolGate(
execute,
**dashboard_approval(DashboardApprovalConfig(api_key="tg_live_...")),
agent_name="deploy-bot",
)
gate.describe("Deploy v2.3.1 to production")
await gate.finalize()
# → Agent pauses. Open the dashboard, review and approve actions.
Inspection
gate.reads # all reads that were executed
gate.pending # all intercepted actions
gate.current_session # full session snapshot
gate.summary() # pretty-printed CLI summary
Dashboard
The ToolGate Dashboard shows all your agent sessions in real time.
Features:
- 🔐 Multi-user auth (email + password)
- 🔑 API key management (generate, revoke, copy)
- 📊 Real-time session monitoring
- ✅ Approve, reject, or edit pending actions
- 🔒 User-scoped data isolation
Getting started:
- Sign up at toolgate.dev
- Go to API Keys → + New Key
- Copy the key into your code — done
Full Example: Autonomous Email Agent
import asyncio
from toolgate import ToolGate, cli_approval
async def execute(tool: str, params: dict) -> any:
tools = {
"listEmails": lambda p: [{"id": 1, "from": "boss@co.com", "subject": "Q3 Report"}],
"readEmail": lambda p: {"id": 1, "body": "Please review the Q3 numbers."},
"sendEmail": lambda p: {"messageId": "real_123"},
"archiveEmail": lambda p: {"success": True},
}
fn = tools.get(tool, lambda p: {"error": "unknown tool"})
return fn(params)
async def main():
gate = ToolGate(
execute,
api_key="tg_live_...",
agent_name="email-assistant",
on_approval_needed=cli_approval,
)
gate.describe("Read inbox, draft replies, archive processed emails")
# Simulate what an autonomous agent would do:
emails = await gate.proxy("listEmails", {}) # ✅ READ — executes
detail = await gate.proxy("readEmail", {"id": 1}) # ✅ READ — executes
sent = await gate.proxy("sendEmail", { # 🛑 SEND — intercepted
"to": "boss@co.com",
"subject": "Re: Q3 Report",
"body": "Reviewed — numbers look solid. Ship it.",
})
archived = await gate.proxy("archiveEmail", {"id": 1}) # 🛑 DELETE — intercepted
# Agent thinks both worked. Now finalize:
await gate.finalize()
# → CLI shows: 2 reads executed, 2 actions pending approval
asyncio.run(main())
API Reference
ToolGate(executor, **kwargs)
| Kwarg | Type | Description |
|---|---|---|
api_key |
str |
Your ToolGate API key (recommended) |
classifier |
(name, params) -> ToolClassification |
Custom classification function |
read_tools |
list[str] |
Always-passthrough tool names |
action_tools |
list[str] |
Always-intercept tool names |
phantom_response |
(tool: ToolCall) -> Any |
Custom phantom result generator |
on_approval_needed |
(req) -> ApprovalResult |
Async callback when agent finishes |
agent_name |
str |
Display name for the agent |
mcp_servers |
list[MCPServerDef] |
MCP server + tool definitions |
Instance Methods
| Method | Returns | Description |
|---|---|---|
.proxy(name, params) |
Any |
The proxied executor to give to your agent |
.wrap_tools(map) |
dict |
Wraps a {name: fn} tool map |
.describe(text) |
self |
Set task description |
.finalize() |
ApprovalRequest |
End session, trigger approval |
.approve_all() |
dict |
Approve + execute all pending actions |
.reject_all() |
None |
Reject all pending actions |
.execute_approval() |
dict |
Execute with per-action decisions |
.summary() |
str |
Pretty-printed session summary |
License
MIT
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file toolgate_ai-1.0.4.tar.gz.
File metadata
- Download URL: toolgate_ai-1.0.4.tar.gz
- Upload date:
- Size: 21.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.14.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
f381125093fb593752aa19f3c19374217817da2876e2213dd4f65d2814564fc7
|
|
| MD5 |
73eb6d137c1d642009fafa89b7ad7827
|
|
| BLAKE2b-256 |
5f1362edd8bc55364219c79ca260150c109ede9710b1cb1a606a0716189f9193
|
File details
Details for the file toolgate_ai-1.0.4-py3-none-any.whl.
File metadata
- Download URL: toolgate_ai-1.0.4-py3-none-any.whl
- Upload date:
- Size: 21.0 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.14.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
e40f967cbb8fcb769ab266474232b431feb61376572ceff95ae4fdf34c835224
|
|
| MD5 |
5dd935f4c8dbcb989097d2c88079cef4
|
|
| BLAKE2b-256 |
b010fbc01da3b3a332c73a9edd762da05271db4b9f6f1ef6e9b8ef1d937e2896
|