Skip to content
Night Brownie Night Brownie
⌘K

Queue

Queue HTTP endpoints for the queue-mediated agent protocol.

Classes:

Functions:

  • get_drain_event –

    Retrieve the drain asyncio.Event from the app state.

  • get_task_queue –

    Retrieve the TaskQueue from app state.

  • queue_complete –

    Store a completed task decision and signal the drain loop.

  • queue_heartbeat –

    Extend the claim window for an in-progress task.

  • queue_next –

    Claim and return the next pending task for the requesting agent.

HeartbeatRequest pydantic-model

Bases: BaseModel

Request body for POST /queue/heartbeat.

Attributes:

Parameters:

  • task_id (str) –
Show JSON schema: 
{
  "description": "Request body for POST /queue/heartbeat.\n\nAttributes:\n    task_id: ID of the task to heartbeat.",
  "properties": {
    "task_id": {
      "title": "Task Id",
      "type": "string"
    }
  },
  "required": [
    "task_id"
  ],
  "title": "HeartbeatRequest",
  "type": "object"
}

Fields:

  • task_id (str)

NextTaskRequest pydantic-model

Bases: BaseModel

Request body for POST /queue/next.

Attributes:

Parameters:

  • agent_url (str) –
Show JSON schema: 
{
  "description": "Request body for POST /queue/next.\n\nAttributes:\n    agent_url: Base URL of the agent requesting a task.",
  "properties": {
    "agent_url": {
      "title": "Agent Url",
      "type": "string"
    }
  },
  "required": [
    "agent_url"
  ],
  "title": "NextTaskRequest",
  "type": "object"
}

Fields:

  • agent_url (str)

get_drain_event 

get_drain_event(request: Request) -> Event | None

Retrieve the drain asyncio.Event from the app state.

Parameters:

  • request (Request) –

    The incoming FastAPI request.

Returns:

  • Event | None –

    The drain asyncio.Event, or None if not yet initialized.

get_task_queue 

get_task_queue(request: Request) -> TaskQueue

Retrieve the TaskQueue from app state.

Parameters:

  • request (Request) –

    The incoming FastAPI request.

Returns:

  • TaskQueue –

    The TaskQueue instance that is attached to the app state.

queue_complete async 

queue_complete(decision: DecisionMessage, task_queue: TaskQueue = Depends(get_task_queue), drain_event: Event | None = Depends(get_drain_event)) -> Response

Store a completed task decision and signal the drain loop.

Parameters:

  • decision (DecisionMessage) –

    The agent's DecisionMessage.

  • task_queue (TaskQueue, default: Depends(get_task_queue) ) –

    Task queue from the app state (injected).

  • drain_event (Event | None, default: Depends(get_drain_event) ) –

    Drain loop event from the app state (injected).

Returns:

  • Response –

    202 Accepted.

queue_heartbeat async 

queue_heartbeat(body: HeartbeatRequest, task_queue: TaskQueue = Depends(get_task_queue)) -> Response

Extend the claim window for an in-progress task.

Parameters:

Returns:

  • Response –

    202 Accepted.

queue_next async 

queue_next(body: NextTaskRequest, task_queue: TaskQueue = Depends(get_task_queue)) -> Response

Claim and return the next pending task for the requesting agent.

Parameters:

Returns:

  • Response –

    200 with TaskMessage JSON when a task is available, 204 when the queue is empty.