Overview
ToolContext carries request-scoped identity to your tools—answering the question “who is making this request?” It provides user IDs, organization IDs, session information, and auth claims that tools need to act on behalf of specific users.
Design principle: ToolContext answers “who is making this request?” not “what infrastructure does this tool need?”Infrastructure dependencies (database clients, API clients, caches) are better captured at tool definition time via closures. This keeps your tools testable and your context lightweight.
Class Definition
from dataclasses import dataclass, field
from typing import Dict, Any, Optional, Callable, Awaitable
@dataclass
class ToolContext:
"""Context passed to tools during execution."""
tool_name: Optional[str] = None # Name of the tool being executed
tool_call_id: Optional[str] = None # Unique ID for this tool call
deps: Dict[str, Any] = field(default_factory=dict) # User dependencies
progress_callback: Optional[Callable[[int, int, str], Awaitable[None]]] = None # For progress updates
Features
- Request identity: Carry user_id, org_id, session_id, and auth claims
- Typed metadata fields: Access
tool_name and tool_call_id directly
- Dict-style access: Simple
ctx["key"] syntax for accessing identity data
- Lightweight: Only request-scoped data, not infrastructure
Using ToolContext
The Recommended Pattern
Close over infrastructure at tool definition time, receive identity at runtime:
from tyler import ToolContext
# Infrastructure is captured when the tool is defined
def create_order_tools(db, payment_api):
"""Create order tools with infrastructure closed over."""
async def get_my_orders(ctx: ToolContext, status: str = "all") -> str:
"""Get orders for the current user."""
# Identity from context - "who is asking?"
user_id = ctx["user_id"]
org_id = ctx.get("org_id")
# Infrastructure from closure - captured at definition time
orders = await db.query(
"SELECT * FROM orders WHERE user_id = ? AND org_id = ?",
[user_id, org_id]
)
return json.dumps(orders)
async def create_order(ctx: ToolContext, product_id: str, quantity: int) -> str:
"""Create an order for the current user."""
user_id = ctx["user_id"]
# Use closed-over infrastructure
order = await db.insert("orders", {
"user_id": user_id,
"product_id": product_id,
"quantity": quantity
})
await payment_api.authorize(user_id, order.total)
return f"Order {order.id} created"
return [get_my_orders, create_order]
# At startup: create tools with infrastructure
db = Database(connection_string)
payment = PaymentAPI(api_key)
order_tools = create_order_tools(db, payment)
agent = Agent(model_name="gpt-4o", tools=order_tools)
# At request time: pass only identity
await agent.run(thread, tool_context={
"user_id": current_user.id,
"org_id": current_user.org_id,
"permissions": current_user.permissions
})
Accessing Identity Data
Tools access request-scoped identity using dict-style syntax:
from tyler import ToolContext
async def get_user_preferences(ctx: ToolContext) -> str:
"""Get preferences for the current user."""
user_id = ctx["user_id"] # Required - raises KeyError if missing
org_id = ctx.get("org_id") # Optional - returns None if missing
session_id = ctx.get("session_id") # Optional
# ... fetch preferences for this user
async def audited_action(ctx: ToolContext, action: str) -> str:
"""Perform an action with audit logging."""
# Request identity
user_id = ctx["user_id"]
# Execution metadata (auto-populated)
tool_name = ctx.tool_name # e.g., "audited_action"
call_id = ctx.tool_call_id # e.g., "call_abc123"
# Log the audit trail
await audit_log.write(
user=user_id,
action=action,
tool=tool_name,
call_id=call_id
)
return f"Completed: {action}"
Passing Context to Agent
Context is provided per-request with the identity of who is making the request:
from tyler import Agent
# Agent setup (tools have infrastructure closed over)
agent = Agent(
model_name="gpt-4o",
tools=order_tools # Already have db, api clients via closures
)
# Each request passes identity
result = await agent.run(
thread,
tool_context={
"user_id": request.user.id,
"org_id": request.user.org_id,
"session_id": request.session_id,
"permissions": request.user.permissions
}
)
Agent-Level Context (Shared Identity Defaults)
For scenarios where some identity data is constant (e.g., service accounts, system agents):
# System agent that always acts as a specific service account
system_agent = Agent(
model_name="gpt-4o",
tools=admin_tools,
tool_context={
"user_id": "system",
"role": "admin"
}
)
# No per-request context needed for system operations
await system_agent.run(thread)
Context Merging
When both agent-level and run-level contexts are provided, they merge with run-level taking precedence:
agent = Agent(
tool_context={"org_id": "default_org", "tier": "free"}
)
# Final context: org_id overridden, tier inherited, user_id added
await agent.run(thread, tool_context={
"org_id": "premium_org", # Overrides agent default
"user_id": "user_123" # Added for this request
})
# Result: {"org_id": "premium_org", "tier": "free", "user_id": "user_123"}
Parameter Naming Convention
The tool runner looks for specific parameter names to identify context parameters:
Preferred short form. Must be the first parameter.async def my_tool(ctx: ToolContext, param: str) -> str:
...
Alternative longer form. Must be the first parameter.async def my_tool(context: ToolContext, param: str) -> str:
...
The context parameter must be the first parameter in your function signature. If it appears elsewhere, it won’t receive the injected context.
Typed Fields
These fields are automatically populated by the agent:
| Field | Type | Description |
|---|
tool_name | str | None | Name of the tool being executed |
tool_call_id | str | None | Unique identifier for this tool call |
deps | Dict[str, Any] | User-provided dependencies |
progress_callback | Callable | None | Async callback for reporting progress (MCP tools) |
async def my_tool(ctx: ToolContext, param: str) -> str:
# Typed field access
print(f"Running: {ctx.tool_name}") # e.g., "my_tool"
print(f"Call ID: {ctx.tool_call_id}") # e.g., "call_abc123"
# Dict access to user deps
db = ctx["db"] # Accesses ctx.deps["db"]
return "done"
Dict-Style Access Methods
ToolContext supports full dict-style access for backward compatibility:
| Method | Example | Description |
|---|
[] | ctx["key"] | Get value, raises KeyError if missing |
[]= | ctx["key"] = val | Set value |
get() | ctx.get("key", default) | Get with default |
in | "key" in ctx | Check key exists |
keys() | ctx.keys() | Iterate over keys |
items() | ctx.items() | Iterate over key-value pairs |
values() | ctx.values() | Iterate over values |
len() | len(ctx) | Count of deps |
Common Context Keys
Here are common keys used in ToolContext—all represent request identity, not infrastructure:
| Key | Type | Description |
|---|
user_id | str | Current user’s ID |
org_id | str | Organization or tenant ID |
session_id | str | Current session identifier |
permissions | list[str] | User’s permissions/scopes |
roles | list[str] | User’s roles |
auth_claims | dict | JWT claims or auth metadata |
request_id | str | Trace ID for logging/debugging |
locale | str | User’s locale preference |
Avoid passing infrastructure in context. Database connections, API clients, caches, and loggers should be closed over when defining tools, not passed per-request.
Examples
Multi-Tenant Data Access
def create_data_tools(db):
"""Tools with database closed over, identity from context."""
async def query_orders(ctx: ToolContext, status: str, limit: int = 10) -> str:
"""Query orders for the current user."""
user_id = ctx["user_id"]
org_id = ctx["org_id"] # Tenant isolation
orders = await db.query(
"SELECT * FROM orders WHERE user_id = ? AND org_id = ? AND status = ? LIMIT ?",
[user_id, org_id, status, limit]
)
return json.dumps([dict(o) for o in orders])
return [query_orders]
Permission-Gated Actions
def create_admin_tools(db, notification_service):
"""Admin tools that check permissions from context."""
async def delete_user(ctx: ToolContext, target_user_id: str) -> str:
"""Delete a user (admin only)."""
permissions = ctx.get("permissions", [])
if "admin:delete_users" not in permissions:
return "Error: Insufficient permissions"
admin_id = ctx["user_id"] # Who is performing the action
await db.delete("users", target_user_id)
await notification_service.notify_admins(
f"User {target_user_id} deleted by {admin_id}"
)
return f"User {target_user_id} deleted"
return [delete_user]
Personalized Recommendations
def create_recommendation_tools(recommender_api):
"""Recommendation tools with API client closed over."""
async def get_recommendations(ctx: ToolContext, category: str) -> str:
"""Get personalized recommendations for the current user."""
user_id = ctx["user_id"]
locale = ctx.get("locale", "en-US")
recs = await recommender_api.get_recommendations(
user_id=user_id,
category=category,
locale=locale
)
return json.dumps(recs)
return [get_recommendations]
Audit Logging
def create_audited_tools(db, audit_log):
"""Tools that log who performed each action."""
async def update_settings(ctx: ToolContext, settings: dict) -> str:
"""Update account settings with audit trail."""
user_id = ctx["user_id"]
session_id = ctx.get("session_id")
request_id = ctx.get("request_id")
await db.update("settings", user_id, settings)
await audit_log.record({
"action": "update_settings",
"user_id": user_id,
"session_id": session_id,
"request_id": request_id,
"changes": settings
})
return "Settings updated"
return [update_settings]
Error Handling
Missing Context
When a tool expects identity but none is provided:
from tyler import ToolContextError
async def requires_user(ctx: ToolContext, action: str) -> str:
user_id = ctx["user_id"] # KeyError if not in context
...
# This raises ToolContextError
try:
result = await agent.run(thread) # No tool_context!
except ToolContextError as e:
print(f"Missing context: {e}")
Missing Identity Keys
Handle optional identity gracefully:
async def flexible_tool(ctx: ToolContext, action: str) -> str:
# Required identity
if "user_id" not in ctx:
raise ValueError("This tool requires user_id in context")
user_id = ctx["user_id"]
# Optional identity with defaults
org_id = ctx.get("org_id", "default")
locale = ctx.get("locale", "en-US")
permissions = ctx.get("permissions", [])
# Check permissions before proceeding
if "write" not in permissions:
return "Error: Write permission required"
# ... rest of implementation
Backward Compatibility
Tools Without Context
Tools without a context parameter work normally:
# This tool doesn't need identity
async def simple_math(a: int, b: int) -> str:
return str(a + b)
# Context is ignored for this tool
await agent.run(
thread,
tool_context={"user_id": "123"} # Passed but not used
)
Existing Code Works Unchanged
The ToolContext dataclass is fully backward compatible. Existing tools using dict-style access continue to work:
# This code works the same before and after the update
async def existing_tool(ctx: ToolContext, query: str) -> str:
user_id = ctx["user_id"] # Still works
org_id = ctx.get("org_id") # Still works
if "permissions" in ctx: # Still works
permissions = ctx["permissions"]
return "done"
Testing with Context
Testing is clean because infrastructure is separate from identity:
import pytest
from unittest.mock import AsyncMock
@pytest.mark.asyncio
async def test_query_orders():
# Mock the infrastructure (closed over at tool creation)
mock_db = AsyncMock()
mock_db.query.return_value = [
{"id": 1, "status": "pending"},
{"id": 2, "status": "pending"}
]
# Create tool with mock infrastructure
tools = create_data_tools(mock_db)
query_orders = tools[0]
# Test with identity context only
ctx = {"user_id": "user_123", "org_id": "org_456"}
result = await query_orders(ctx, status="pending", limit=10)
# Verify
assert "pending" in result
mock_db.query.assert_called_once()
# Verify user_id was used in query
call_args = mock_db.query.call_args[0]
assert "user_123" in call_args[1]
assert "org_456" in call_args[1]
Testing Permission Checks
@pytest.mark.asyncio
async def test_admin_action_requires_permission():
mock_db = AsyncMock()
tools = create_admin_tools(mock_db, AsyncMock())
delete_user = tools[0]
# Test without admin permission
ctx = {"user_id": "user_123", "permissions": ["read"]}
result = await delete_user(ctx, target_user_id="user_456")
assert "Insufficient permissions" in result
mock_db.delete.assert_not_called()
# Test with admin permission
ctx = {"user_id": "admin_1", "permissions": ["admin:delete_users"]}
result = await delete_user(ctx, target_user_id="user_456")
assert "deleted" in result
mock_db.delete.assert_called_once()
Best Practices
Close Over Infrastructure
Capture databases, API clients, and other infrastructure when defining tools:
# ✅ Good: Infrastructure closed over at definition time
def create_tools(db, cache, external_api):
async def fetch_data(ctx: ToolContext, query: str) -> str:
user_id = ctx["user_id"] # Identity from context
return await db.query(query, user_id) # Infrastructure from closure
return [fetch_data]
# ❌ Avoid: Infrastructure in context
async def fetch_data(ctx: ToolContext, query: str) -> str:
db = ctx["db"] # Infrastructure shouldn't be here
return await db.query(query, ctx["user_id"])
Document Expected Identity
Document what identity keys your tools expect:
async def sensitive_action(ctx: ToolContext, action: str) -> str:
"""
Perform a sensitive action.
Args:
ctx: Request context containing:
- user_id (str): Required. The authenticated user's ID.
- org_id (str): Required. The user's organization.
- permissions (list[str]): Required. User's permission scopes.
- session_id (str): Optional. For audit logging.
action: The action to perform.
Returns:
Result of the action.
"""
Validation Helper
Create a validation helper for required identity:
def require_identity(ctx: ToolContext, *keys: str) -> None:
"""Validate that context contains required identity keys."""
missing = [key for key in keys if key not in ctx]
if missing:
raise ValueError(f"Missing required identity: {missing}")
async def my_tool(ctx: ToolContext, param: str) -> str:
require_identity(ctx, "user_id", "org_id")
...
Identity Factory
Create identity context consistently from your auth layer:
class IdentityContext:
"""Build tool context from authenticated requests."""
@staticmethod
def from_request(request) -> dict:
"""Extract identity from an HTTP request."""
return {
"user_id": request.user.id,
"org_id": request.user.org_id,
"permissions": request.user.permissions,
"session_id": request.session.id,
"request_id": request.headers.get("X-Request-ID"),
"locale": request.headers.get("Accept-Language", "en-US")
}
@staticmethod
def from_jwt(claims: dict) -> dict:
"""Extract identity from JWT claims."""
return {
"user_id": claims["sub"],
"org_id": claims.get("org"),
"permissions": claims.get("scope", "").split(),
"roles": claims.get("roles", [])
}
# Usage in a web framework
@app.post("/chat")
async def chat(request: Request, message: str):
identity = IdentityContext.from_request(request)
result = await agent.run(thread, tool_context=identity)
return result
See Also
