Streaming

Consume Server-Sent Events from chat and pentest execution, follow agent activity, and attach files to a streaming request.

Real-time work in Rank is delivered over Server-Sent Events (SSE). Two methods open a stream:

client.ai.chat.stream(...) — send a prompt to an agent, or launch a pentest, and stream the response as it is produced.
client.pentests.stream(pentest_id) — reconnect to a pentest that is already running.

Both return a Stream on the synchronous client and an AsyncStream on the asynchronous one. The object is both a context manager and an iterator of ServerSentEvent, and it closes itself automatically when a terminal event (complete, error, cancelled or control_stop) arrives.

Sending a chat request

client.ai.chat.stream has a single required argument, user_prompt. The remaining arguments select the flow:

General chat — pass agent_id (and optionally chat_id to persist context).
Pentest, automatic — pass pentest_id with mode="automatic".
Pentest, guided — pass pentest_id with mode="guided" and phase_id.

Argument	Type	Notes
`user_prompt`	`str`	The message to send. Required.
`agent_id`	`int`	Agent to use (required for general chat).
`agent_type`	`"general"` \| `"pentest"`	Inferred automatically when `pentest_id` is set.
`chat_id`	`int`	Persists messages and conversation context across modes.
`pentest_id`	`int`	Required for pentest modes.
`mode`	`"guided"` \| `"automatic"`	Pentest only.
`phase_id`	`int`	Required for guided mode.
`context`	`list`	Explicit conversation context (`ChatMessage` objects or dicts).
`file` / `files`	attachment(s)	A single attachment, or a list. Mutually exclusive.

Sync vs. async iteration

The synchronous client uses a plain with / for. The asynchronous client requires you to await the stream(...) call first, then use async with / async for.

import rank

with rank.Rank() as client:
    with client.ai.chat.stream(
        agent_id=23,
        user_prompt="Summarize the latest critical Apache CVEs disclosed this month.",
        chat_id=14,
    ) as stream:
        for event in stream:
            if event.type == "content":
                print(event.content, end="", flush=True)
            elif event.type == "complete":
                print()

import asyncio
import rank

async def main():
    async with rank.AsyncRank() as client:
        async with await client.ai.chat.stream(
            agent_id=23,
            user_prompt="Summarize the latest critical Apache CVEs disclosed this month.",
            chat_id=14,
        ) as stream:
            async for event in stream:
                if event.type == "content":
                    print(event.content, end="", flush=True)
                elif event.type == "complete":
                    print()

asyncio.run(main())

The ServerSentEvent shape

Every iteration yields a ServerSentEvent. These are the properties you read:

Property	Type	Description
`event.type`	`str`	Event type identifier (`content`, `complete`, `agent_event`, …).
`event.content`	`str`	Text content of the event.
`event.error`	`str \| None`	Error message (only for `error` events).
`event.metadata`	`dict`	Extra fields: progress %, vulnerability counts, elapsed time, etc.
`event.timestamp`	`float \| None`	Event timestamp.
`event.raw_data`	`str`	Raw JSON string before parsing.
`event.is_agent_event`	`bool`	`True` when the event carries an `AgentEvent` payload.
`event.agent_event`	`AgentEvent \| None`	Parsed `AgentEvent` (only when `type == "agent_event"`).
`event.event_type`	`str \| None`	Shortcut to `agent_event.event_type`.

Common transport event types:

`type`	Emitted when
`content`	A chunk of the assistant’s answer.
`queued`	The pentest is waiting for a free execution slot.
`ready`	A slot opened and the run is starting.
`phase_start`	An automatic-mode phase began.
`phase_complete`	A phase finished.
`phase_retry`	A phase is being retried.
`processing_vulnerabilities`	Findings are being analyzed at the end of an automatic run.
`vulnerabilities_complete`	Findings have been stored.
`agent_event`	Live agent activity (thinking, tool calls, orchestration).
`complete`	The stream finished successfully (terminal).
`error`	Something failed; carries an `error` message (terminal).
`cancelled`	The run was cancelled (terminal).
`reconnected`	First event when re-attaching to a running pentest.
`control_stop`	The run was halted by engagement control — kill, closed window, or no active RoE.
`window_warning`	The authorised testing window is about to close (T-60 / T-30 / T-5 minutes).

The golden rule: content vs. agent_event

This is the single most important thing to get right when rendering a stream:

Only event.type == "content" builds the assistant’s answer. Events with event.type == "agent_event" describe what the agent is doing — thinking, calling tools, spawning sub-agents — and are meant for an activity / timeline view. Never concatenate agent_event text into the answer body.

If you only care about the final answer, handle content and ignore everything else. To also render the live activity, branch on event.is_agent_event and read the typed payload from event.agent_event.

import rank
from rank import AgentEvent

with rank.Rank() as client:
    with client.ai.chat.stream(
        agent_id=23,
        user_prompt="Investigate this target end-to-end and summarize the exposure.",
        chat_id=14,
    ) as stream:
        for event in stream:
            if event.type == "content":
                # The actual answer — only this builds the message body.
                print(event.content, end="", flush=True)

            elif event.is_agent_event:
                ev = event.agent_event
                indent = "    " if ev.depth > 0 else ""  # sub-agents are nested
                if ev.event_type == AgentEvent.THINKING:
                    print(f"\n{indent}[thinking] {ev.data['content'][:80]}")
                elif ev.event_type == AgentEvent.TOOL_CALL:
                    print(f"\n{indent}[tool] {ev.data['tool_name']}")
                elif ev.event_type == AgentEvent.AGENT_FINISHED:
                    print(f"\n{indent}[done] {ev.data.get('stop_reason')}")

            elif event.type == "complete":
                print("\n[stream finished]")

import asyncio
import rank
from rank import AgentEvent

async def main():
    async with rank.AsyncRank() as client:
        async with await client.ai.chat.stream(
            agent_id=23,
            user_prompt="Investigate this target end-to-end and summarize the exposure.",
            chat_id=14,
        ) as stream:
            async for event in stream:
                if event.type == "content":
                    print(event.content, end="", flush=True)

                elif event.is_agent_event:
                    ev = event.agent_event
                    indent = "    " if ev.depth > 0 else ""
                    if ev.event_type == AgentEvent.TOOL_CALL:
                        print(f"\n{indent}[tool] {ev.data['tool_name']}")
                    elif ev.event_type == AgentEvent.AGENT_FINISHED:
                        print(f"\n{indent}[done] {ev.data.get('stop_reason')}")

                elif event.type == "complete":
                    print("\n[stream finished]")

asyncio.run(main())

The AgentEvent payload

When event.is_agent_event is True, event.agent_event returns a typed AgentEvent:

Field	Type	Description
`event_type`	`str`	Sub-event identifier (e.g. `thinking`, `tool_call`, `agent_finished`).
`agent_id`	`int \| str \| None`	Numeric agent ID, `"orchestrator"`, or `None`.
`parent_agent_id`	`int \| None`	Parent agent ID for sub-agents.
`instance_id`	`str`	Unique per-execution key — group an instance’s events by this.
`depth`	`int`	`0` = top-level agent, `1` = sub-agent.
`iteration`	`int`	Current iteration of the agent loop.
`timestamp`	`float`	Unix timestamp.
`data`	`dict`	Event-specific payload (read the keys relevant to `event_type`).

Every sub-event name is available as a constant on the class — AgentEvent.TOOL_CALL, AgentEvent.THINKING, AgentEvent.SUBAGENT_SPAWN, AgentEvent.AGENT_FINISHED, and so on — so you can compare against constants instead of raw strings. Sub-agent text arrives as agent_event with event_type == "text_chunk" and depth > 0: it is activity, not the answer.

Full agent_event reference

The harness emits one agent_event per step. You only need to handle the ones your UI cares about; unknown event_type values are safe to ignore. For the mechanics behind these events, see Agent harness & long-running tasks.

Agent loop (a single agent working through its ReAct loop):

`event_type`	Constant	Key `data` fields
`agent_start`	`AGENT_START`	`model`, `agent_id`, `mission`, `max_iterations`, `tools`
`plan`	`PLAN`	`plan_text`
`iteration_start`	`ITERATION_START`	`iteration`, `max_iterations`, `tokens_used`, `stagnation`
`thinking`	`THINKING`	`content`, `streaming`
`tool_call`	`TOOL_CALL`	`tool_name`, `tool_args`
`tool_result`	`TOOL_RESULT`	`tool_name`, `result`, `duration_ms`, `skipped`, `cached`, `blocked`
`interpretation`	`INTERPRETATION`	`content`
`context_compaction`	`CONTEXT_COMPACTION`	`summarized_steps`, `summary_chars`
`shared_context`	`SHARED_CONTEXT`	`summary_length`, `summary_preview`
`nudge`	`NUDGE`	`nudge_count`, `unused_tools`
`progress`	`PROGRESS`	`iteration`, `tools_used`, `tokens_used`
`iteration_complete`	`ITERATION_COMPLETE`	`iteration_prompt`, `iteration_response`, token counts
`text_chunk`	`TEXT_CHUNK`	`content` (sub-agents only, `depth > 0`)
`subagent_spawn`	`SUBAGENT_SPAWN`	`subagent_id`, `mission`, `depth`
`subagent_complete`	`SUBAGENT_COMPLETE`	`subagent_id`, `result_preview`, `iterations`
`agent_finished`	`AGENT_FINISHED`	`stop_reason`, `iterations`, `input_tokens`, `output_tokens`, `elapsed_s`

Orchestration (automatic pentests, where agent_id is "orchestrator"):

`event_type`	Constant	Key `data` fields
`orchestration_start`	`ORCHESTRATION_START`	`agents`, `phase`, `total_agents`
`orchestration_status`	`ORCHESTRATION_STATUS`	`alive_agents`, `completed_agents`, `elapsed_time`
`agent_status_change`	`AGENT_STATUS_CHANGE`	`agent_id`, `status`, `reasoning`
`consolidation_start`	`CONSOLIDATION_START`	`agents_completed`
`consolidation_heartbeat`	`CONSOLIDATION_HEARTBEAT`	`elapsed_seconds`, `message`
`consolidation_complete`	`CONSOLIDATION_COMPLETE`	`input_tokens`, `output_tokens`
`orchestration_complete`	`ORCHESTRATION_COMPLETE`	`phase`, `total_agents`, `successful`, `failed`
`orchestration_cancelled`	`ORCHESTRATION_CANCELLED`	`message`

Browser agent (interactive web navigation): browser_agent_start (BROWSER_AGENT_START) with mission, target_url, max_iterations. Its tool_call/tool_result events mask credentials.

The stop_reason carried by agent_finished is one of goal_reached, max_iterations, timeout, budget, stagnation or cancelled, each available as a constant (AgentEvent.STOP_GOAL_REACHED, AgentEvent.STOP_MAX_ITERATIONS, and so on).

Launching and reconnecting to a pentest

A pentest stream carries the same content and agent_event items plus pentest-specific transport events such as phase_start, phase_complete and vulnerabilities_complete. Launch one with client.ai.chat.stream(..., pentest_id=...), and reattach to a running one with client.pentests.stream(pentest_id).

Launching without an active Rules of Engagement fails. The streaming backend returns 409 Conflict (no_active_roe). Activate a RoE that covers every asset first — client.pentests.roe.create(..., activate=True) — see Resources. An active kill switch is 403 Forbidden; a killed pentest cannot resume. Window close and kill also arrive in-stream as window_warning (T-60 / T-30 / T-5) and control_stop.

import rank

with rank.Rank() as client:
    with client.pentests.stream(pentest_id=42) as stream:
        for event in stream:
            if event.type == "content":
                print(event.content, end="", flush=True)
            elif event.is_agent_event:
                ev = event.agent_event
                print(f"\n{ev.event_type}: {ev.data}")
            elif event.type == "phase_complete":
                print(f"\n[phase {event.metadata.get('phase_id')} done]")
            elif event.type == "complete":
                print("\nPentest finished!")

import asyncio
import rank

async def main():
    async with rank.AsyncRank() as client:
        async with await client.pentests.stream(pentest_id=42) as stream:
            async for event in stream:
                if event.type == "content":
                    print(event.content, end="", flush=True)
                elif event.is_agent_event:
                    ev = event.agent_event
                    print(f"\n{ev.event_type}: {ev.data}")
                elif event.type == "complete":
                    print("\nPentest finished!")

asyncio.run(main())

Attachments

Pass a single attachment with file or several with files (the two are mutually exclusive). Each attachment can be a file-like object, raw bytes, a (filename, content) tuple, or a (filename, content, content_type) tuple. Supported extensions are PDF, JSON, PNG, JPEG, WEBP and GIF, with a limit of 10 files, 30 MB per file and 50 MB combined per request.

import rank

with rank.Rank() as client:
    # Single attachment
    with open("report.pdf", "rb") as f:
        with client.ai.chat.stream(
            agent_id=23,
            user_prompt="Analyze this security report and summarize the findings",
            file=f,
        ) as stream:
            for event in stream:
                if event.type == "content":
                    print(event.content, end="", flush=True)

    # Multiple attachments
    with open("report.pdf", "rb") as a, open("diagram.png", "rb") as b:
        with client.ai.chat.stream(
            agent_id=23,
            user_prompt="Cross-reference these files and summarize the findings",
            files=[a, b],
        ) as stream:
            for event in stream:
                if event.type == "content":
                    print(event.content, end="", flush=True)

Once you can read a stream, see Errors & pagination for handling failures, or the Cookbook for a full streaming UI recipe.