Skip to content

Message queue (steering)

Talk to an agent while it's still working. No cancel, no restart — you push a message into the running loop and the agent picks it up at the next natural moment.

This is what powers "stop, try something else" and "when you're done, also do X" — without throwing away in-flight tool results or busting the prompt cache.

The two moments you can speak

There are exactly two points where it's safe to slip a message in, and you choose which one:

You want to… Use Delivered
Redirect the agent mid-task steering before the next model call, after the current tool batch finishes
Add work for when it's wrapping up follow-up when the agent would otherwise stop

Steering is an interrupt: "actually, do it this way instead." Follow-up is a queue: "once that's done, here's the next thing."

Show it

Python
import asyncio
from pydantic_deep import create_deep_agent, DeepAgentDeps, StateBackend
from pydantic_deep.features.message_queue import MessageQueue, run_with_queue


async def main():
    queue = MessageQueue()
    agent = create_deep_agent(
        model="anthropic:claude-sonnet-4-6",
        message_queue=queue,
    )
    deps = DeepAgentDeps(backend=StateBackend(), message_queue=queue)

    async def supervisor():
        # Some other coroutine, watching the run unfold.
        await asyncio.sleep(2)
        await queue.steer("stop digging deeper — summarise what you have so far")
        await queue.follow_up("when you're done, write one test for the result")

    _, result = await asyncio.gather(
        supervisor(),
        run_with_queue(agent, "investigate the failing test", deps=deps, queue=queue),
    )
    print(result.output)


asyncio.run(main())

Run it

The agent starts investigating. Two seconds in, supervisor() drops a steering message and a follow-up. At the next model call the agent sees [steering] stop digging deeper… and changes course. When it finishes that work, run_with_queue notices the pending follow-up and re-enters the loop with "when you're done, write one test…" as the next prompt.

You steered a running agent from outside it — and queued a continuation it'll reach on its own.

Check it

Comment out the queue.steer(...) line and run again. Without the nudge the agent keeps investigating on its own terms. The queue is the only thing that changed its mind.

Step by step

Step 1: one queue, shared two ways

Python
queue = MessageQueue()
agent = create_deep_agent(model="anthropic:claude-sonnet-4-6", message_queue=queue)
deps = DeepAgentDeps(backend=StateBackend(), message_queue=queue)

The same MessageQueue goes to both the agent and the deps. Passing it to create_deep_agent(message_queue=…) installs the capability that injects steering before each model request. Putting it on DeepAgentDeps makes it reachable from inside tools and subagents (more on that below).

Step 2: speak from anywhere

Python
await queue.steer("change your approach")
await queue.follow_up("then summarise")

Both are async and asyncio-safe — call them from any coroutine, task, or HTTP handler that holds the queue. steer() is the interrupt; follow_up() is the continuation. Each takes a plain string.

Step 3: run with run_with_queue

Python
result = await run_with_queue(agent, "investigate the bug", deps=deps, queue=queue)

Steering works on a plain await agent.run(...) too — the capability injects it before each model request automatically. But follow-ups need someone to re-enter the loop once a run ends. That's run_with_queue: after each completed run it drains any follow-ups and starts a fresh turn with them as the prompt, repeating until the queue is empty.

Use the wrapper for follow-ups

Steering only? Plain agent.run(..., deps=deps) is enough. The moment you want follow-ups, reach for run_with_queue — it's the part that loops.

One message or the whole batch

Every message carries a delivery_mode. The default, "one_at_a_time", hands the agent exactly one queued message per drain — drip-fed, in order. Mark a message "all" and the queue empties the contiguous run of "all" messages at once.

Python
await queue.steer("first hint")
await queue.steer("second hint")
# the agent sees "first hint" now; "second hint" waits for the next model call

await queue.steer("context line 1", delivery_mode="all")
await queue.steer("context line 2", delivery_mode="all")
# both arrive together in one steering injection

The mode is read from the head message, and an "all" head never swallows a later message that asked to be delivered on its own — so you can mix the two freely.

Subagents can steer the parent

Because the queue lives on DeepAgentDeps, anything with ctx.deps can use it — including a subagent. By default clone_for_subagent() hands subagents the same queue, so a child can talk back to the parent:

Python
# Inside a tool or subagent:
await ctx.deps.message_queue.steer("parent — this path is a dead end, pivot")

Want isolation instead? Give the cloned deps a fresh MessageQueue() and the channels stay separate.

It's a back-channel, not a megaphone

Steering injects a single user-turn part before the next request. Keep each message short and directive — the agent reads it like a teammate leaning over your shoulder, not a new system prompt.

Recap

  • A MessageQueue lets external code — or a subagent — inject messages into a running agent without cancel/restart.
  • Steering redirects mid-task (before the next model call); follow-up adds work for when the agent would stop.
  • Share one queue with both create_deep_agent(message_queue=…) and DeepAgentDeps(message_queue=…).
  • Steering works on plain agent.run; follow-ups need run_with_queue to re-enter the loop.
  • delivery_mode="all" batches messages; the default drip-feeds one at a time.

Where to go next: