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")

Parameters:

Name	Type	Description	Default
max_pending	int | None	Cap on pending messages per priority. None removes the cap. Enqueueing past the cap raises :class:QueueFullError.	DEFAULT_MAX_PENDING

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

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

Raises:

Type	Description
QueueFullError	When max_pending steering messages already wait.

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

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

Raises:

Type	Description
QueueFullError	When max_pending follow-up messages already wait.

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.

discard_follow_up(*, keep=None) async

Remove every pending follow-up and return the removed ones.

Messages for which keep returns True stay queued, in order. Callers use this to prune follow-ups that a cancelled run made stale while sparing the ones no longer tied to it — an externally submitted message was never about the cancelled task, and dropping it reads to its sender as being ignored.

Parameters:

Name	Type	Description	Default
keep	Callable[[QueuedMessage], bool] | None	Predicate deciding which pending messages survive. None discards all of them.	None

Returns:

Type	Description
list[QueuedMessage]	The messages removed from the queue, in submission order.

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.

queued_source

pydantic_deep.features.message_queue.queued_source(message)

Return the metadata["source"] label of message, or None when it has none.

The value is written by whoever enqueued the message and is interpolated into the label the model reads, so it is reduced to word characters plus ., : and - and truncated. A message with no source was submitted locally.

QueueFullError

pydantic_deep.features.message_queue.QueueFullError

Bases: RuntimeError

Raised when a message cannot be queued because that priority is at capacity.

Enqueueing fails loudly rather than dropping: an external bridge needs to know a submission did not land so it can report that back to whoever sent it.