Message Queue API

The message queue injects messages into a running agent — steering (delivered before the next model request) or follow-up (delivered when the agent would otherwise stop). Pass one via message_queue= on create_deep_agent. See Message queue (steering) for the overview.

MessageQueue

pydantic_deep.features.message_queue.MessageQueue

Asyncio-safe dual-priority queue for mid-run message delivery.

Usage::

Text Only

queue = MessageQueue()
agent = create_deep_agent(message_queue=queue)

# From another coroutine / task while the agent is running:
await queue.steer("stop that, try a different approach")
await queue.follow_up("when done, also summarise")

steer(content, *, delivery_mode='one_at_a_time', metadata=None) async

Queue a steering message (delivered before the next LLM call).

follow_up(content, *, delivery_mode='one_at_a_time', metadata=None) async

Queue a follow-up message (delivered when the agent would otherwise stop).

drain_steering() async

Return queued steering messages and remove them from the queue.

drain_follow_up() async

Return queued follow-up messages and remove them from the queue.

has_follow_up()

Return True if there are pending follow-up messages (without the lock, display only).

pending_count()

Return (steering_count, follow_up_count) without the lock (display only).

MessageQueueCapability

pydantic_deep.features.message_queue.MessageQueueCapability dataclass

Bases: AbstractCapability[DeepAgentDeps]

Injects queued steering messages before each model request.

Register via create_deep_agent(message_queue=queue) or directly::

Text Only

capability = MessageQueueCapability(queue=queue)
agent = Agent(model, capabilities=[capability])

run_with_queue

pydantic_deep.features.message_queue.run_with_queue(agent, prompt, *, deps, queue, message_history=None, **run_kwargs) async

Run an agent and re-enter the loop when follow-up messages arrive.

Steering messages are handled automatically by :class:MessageQueueCapability (injected before each LLM request). Follow-up messages are consumed here after each completed run, before the next iteration starts.

Parameters:

Name	Type	Description	Default
agent	Agent[Any, Any]	The agent to run.	required
prompt	str	The initial user prompt.	required
deps	Any	Agent dependencies.	required
queue	MessageQueue	The shared MessageQueue instance.	required
message_history	list[Any] | None	Existing conversation history (passed through to agent.run).	None
**run_kwargs	Any	Extra keyword arguments forwarded to agent.run().	{}

Returns:

Type	Description
Any	The final :class:~pydantic_ai.AgentRunResult.