ToolContext

Request-scoped identity and metadata for tool execution

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

Accessing Metadata Fields

Tools can also access typed metadata about the current execution:

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:

ctx · ToolContext

Preferred short form. Must be the first parameter.

async def my_tool(ctx: ToolContext, param: str) -> str:
    ...

context · ToolContext

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

• Agent - Agent.run() with tool_context
• Adding Tools - Creating custom tools
• Structured Output Guide - Complete guide including tool context