Tool Call Executor Timeout

Cap each tool call in milliseconds and short-circuit pending calls with a cancel token

The tool-call executor caps each tool call with a millisecond timeout and lets you cancel pending calls, returning a typed result the model can reason about instead of hanging the turn.

from praisonaiagents import Agent

def fetch_url(url: str) -> str:
    ...  # a tool that may hang

Agent(
    instructions="Look things up.",
    tools=[fetch_url],
    llm={"model": "gpt-4o", "tool_timeout_ms": 5000},  # 5-second per-tool cap
).start("Look up https://example.com")

Quick Start

1. Cap every tool call

   Set tool_timeout_ms inside the llm= dict. A slow tool now returns a typed timeout result instead of freezing the turn.

   from praisonaiagents import Agent
   
   def fetch_url(url: str) -> str:
       ...  # a tool that may hang
   
   agent = Agent(
       instructions="Fetch pages and summarise them.",
       tools=[fetch_url],
       llm={"model": "gpt-4o", "tool_timeout_ms": 5000},  # 5s per tool
   )
   
   agent.start("Summarise https://example.com")

2. Cancel from another thread

   Signal a threading.Event-shaped cancel token to abort pending tool calls — useful in a REPL or gateway where a user hits stop.

   import threading
   from praisonaiagents.tools import (
       SequentialToolCallExecutor, ToolCall, ToolCancelledError,
   )
   
   cancel = threading.Event()
   
   def execute_tool(name, args, tool_call_id):
       return {"ok": True}
   
   executor = SequentialToolCallExecutor()
   cancel.set()  # e.g. fired from a signal handler / stop button
   
   results = executor.execute_batch(
       [ToolCall("fetch_url", {"url": "https://example.com"}, "call_1")],
       execute_tool,
       cancel_token=cancel,
   )
   
   print(results[0].error_kind)         # "cancelled"
   print(results[0].structured_error)   # {"error": True, "kind": "cancelled", ...}

How It Works

The executor wraps each tool call, branching to a typed result on timeout or cancellation.

A timed call lands in one of three terminal states: the worker finished (success), the deadline hit and a slot was charged lazily (timeout), or the orphan pool is saturated and the call never ran (capacity).

Configuration Options

The cancel token is duck-typed — it supports both threading.Event and InterruptController-shaped tokens without importing a concrete type.

Structured Error Envelope

ToolResult.structured_error returns a discriminated JSON envelope so the model can tell a timeout from a genuine bug, instead of an opaque "Error executing tool: ..." string.

{
  "error": true,
  "kind": "timeout",
  "type": "ToolTimeoutError",
  "message": "Tool 'fetch_url' timed out after 5000ms",
  "tool": "fetch_url"
}

A capacity kind carries ToolResult.result == {"error": "capacity", "tool": "<name>"}, error_kind == "capacity", and ToolResult.error is a ToolCapacityError (“too many hung tool workers; refusing to abandon another”):

{
  "error": true,
  "kind": "capacity",
  "type": "ToolCapacityError",
  "message": "too many hung tool workers; refusing to abandon another",
  "tool": "fetch_url"
}

The kind discriminant lets the model retry a timeout, abort on a cancelled or capacity, and report a plain error differently.

Exceptions

from praisonaiagents.tools import ToolTimeoutError, ToolCancelledError, ToolCapacityError

These are surfaced as ToolResult.error with the matching error_kind — they are not thrown to your calling code.

Common Patterns

Bound a whole batch with one timeout — most users’ entry point:

from praisonaiagents import Agent

def fetch_url(url: str) -> str:
    ...

Agent(
    instructions="Summarise pages.",
    tools=[fetch_url],
    llm={"model": "gpt-4o", "tool_timeout_ms": 3000},  # 3s cap on every tool call
).start("Summarise https://example.com")

Wire up graceful cancellation for a REPL or long-running gateway:

import threading
from praisonaiagents.tools import SequentialToolCallExecutor, ToolCall

cancel = threading.Event()

def stop():
    cancel.set()  # call from a signal handler or a UI stop button

def execute_tool(name, args, tool_call_id):
    return {"ok": True}

executor = SequentialToolCallExecutor()
results = executor.execute_batch(
    [ToolCall("fetch_url", {"url": "https://example.com"}, "call_1")],
    execute_tool,
    timeout_ms=5000,
    cancel_token=cancel,
)

Best Practices

Related