Deferred & Progress Tools

Let slow tools stream progress or defer a result so the turn never freezes

Slow tools can stream progress or hand back a “resolve later” handle so a single long-running call never freezes the whole turn.

Quick Start

1. Stream progress from a tool

   Add an on_progress=None parameter. The executor detects it and wires updates through automatically — no other changes needed.

   from praisonaiagents import Agent
   from praisonaiagents.tools import ToolProgress
   
   def deep_research(topic: str, on_progress=None) -> str:
       if on_progress:
           on_progress(ToolProgress("searching sources..."))
           on_progress(ToolProgress("summarising..."))
       return f"Report on {topic}"
   
   agent = Agent(
       name="Researcher",
       instructions="Answer research questions.",
       tools=[deep_research],
   )
   agent.start("Research quantum computing trends")

2. Defer a long-running job

   Return defer(...) and the model sees the note immediately — no blocking on a 10-minute render.

   from praisonaiagents import Agent
   from praisonaiagents.tools import defer, DeferredToolResult
   
   render_queue = {}
   
   def render_video(script: str) -> DeferredToolResult:
       job_id = "job-123"
       render_queue[job_id] = script
       return defer(
           note=f"Started render job {job_id}. I'll post the video when it's ready.",
           handle_id=job_id,
       )
   
   agent = Agent(name="Studio", instructions="Render videos.", tools=[render_video])
   agent.start("Render the intro sequence")

   Later, call resolve_deferred(job_id, video_url) from the render-complete callback to deliver the result — see Resolving a Deferred Result.

How It Works

The executor inspects each tool’s signature, forwards progress it emits, and records a deferred handle without blocking.

Which return type do I choose?

Pick the simplest option that fits how long your tool runs.

Configuration Options

ToolProgress describes a single incremental update a tool emits while working.

DeferredToolResult is a handle a tool returns when it kicks off background work.

The defer() factory builds a DeferredToolResult; handle_id defaults to a generated uuid.uuid4().hex when omitted.

The enriched ToolResult carries these extra fields alongside result.

The praisonaiagents.tools public deferred-resolver API delivers a background result back into the conversation once the work finishes.

Native OpenAI-SDK parity is available as of the PraisonAI release that includes the #3967 fix and later.

Resolving a Deferred Result

Call resolve_deferred(handle_id, value) when the background job finishes — the value is re-injected into the same conversation as a tool response.

A tool returns defer(...) immediately; a background thread calls resolve_deferred(...) with the same handle_id when it is done.

import threading
from praisonaiagents import Agent
from praisonaiagents.tools import defer, DeferredToolResult, resolve_deferred

def _run_render(job_id: str) -> None:
    video_url = f"https://cdn.example.com/{job_id}.mp4"
    resolve_deferred(job_id, video_url)

def render_video(script: str) -> DeferredToolResult:
    job_id = "job-123"
    threading.Thread(target=_run_render, args=(job_id,)).start()
    return defer(note=f"Rendering {job_id}; I'll post the link when it's ready.", handle_id=job_id)

agent = Agent(name="Studio", instructions="Render videos.", tools=[render_video])
agent.start("Render the intro sequence")

The run loop registers the handle automatically on both the LiteLLM path and the native OpenAI-SDK path, so tool authors only call resolve_deferred(...) from the completion callback — no LLM wiring needed.

Inspect agent.chat_history to confirm the resolved value landed on the durable transcript:

assert any("[deferred:" in (m.get("content") or "") for m in agent.chat_history)

The two-turn cycle: turn 1 the tool returns defer(...); a background thread calls resolve_deferred(...) while the user is idle; turn 2 the user asks a follow-up and the agent replies using the resolved value now present in agent.chat_history.

Deferred results under switch_model()

Swapping the model mid-conversation keeps deferred resolution wired — switch_model(...) re-attaches the Agent’s history callback to the new LLM instance.

from praisonaiagents import Agent
from praisonaiagents.tools import defer, DeferredToolResult, resolve_deferred

def render_video(script: str) -> DeferredToolResult:
    return defer(note="Rendering; I'll post the link when ready.", handle_id="job-123")

agent = Agent(name="Studio", instructions="Render videos.", tools=[render_video])
agent.chat("Render the intro sequence")

agent.switch_model("openai/gpt-4o-mini")   # re-wires the deferred callback
resolve_deferred("job-123", "https://cdn.example.com/job-123.mp4")

# Lands on agent.chat_history, visible on the next turn:
assert any("[deferred:" in (m.get("content") or "") for m in agent.chat_history)
agent.chat("Post the finished video link.")

When does the resolver fire?

The resolver handles the job finishing after registration, before registration (buffered), or being cancelled.

The same sequence applies whether the agent runs on the LiteLLM loop or the native OpenAI-SDK loop.

Common Patterns

Async tool with progress

An async def tool is awaited natively — no asyncio.run wrapper needed.

import asyncio
from praisonaiagents import Agent
from praisonaiagents.tools import ToolProgress

async def deep_research(topic: str, on_progress=None) -> str:
    if on_progress:
        on_progress(ToolProgress("searching..."))
    await asyncio.sleep(0)
    if on_progress:
        on_progress(ToolProgress("summarising..."))
    return f"Report on {topic}"

agent = Agent(name="Researcher", instructions="Research topics.", tools=[deep_research])
agent.start("Research battery chemistry breakthroughs")

Deferred job resolved by handle

Return defer(...) now, then resolve the job later by its handle_id.

from praisonaiagents import Agent
from praisonaiagents.tools import defer, DeferredToolResult, resolve_deferred

jobs = {}

def enqueue_report(topic: str) -> DeferredToolResult:
    handle = defer(note=f"Building report on {topic}.", handle_id=f"report-{topic}")
    jobs[handle.handle_id] = {"topic": topic, "status": "running"}
    return handle

def on_report_done(handle_id: str, content: str) -> None:
    # Re-injects `content` into the same conversation as a tool response.
    resolve_deferred(handle_id, content)

agent = Agent(name="Analyst", instructions="Build reports.", tools=[enqueue_report])
agent.start("Build a market report on EVs")

Structured error inspection

Read structured_error to get the error type and message instead of a flattened string.

from praisonaiagents.tools.call_executor import (
    ToolCall,
    SequentialToolCallExecutor,
)

def execute(name, args, cid):
    raise ValueError("boom")

result = SequentialToolCallExecutor().execute_batch(
    [ToolCall(function_name="risky", arguments={}, tool_call_id="id-1")],
    execute,
)[0]

print(result.structured_error)
# {"error": True, "type": "ValueError", "message": "boom", "tool": "risky"}

Best Practices

Related