ctx.stream

Class: StreamEmitter

class StreamEmitter:
    def emit(self, event_type: str, data: dict | str) -> None: ...
    def progress(self, content: str, *, tool_name: str | None = None) -> None: ...
    def intent(self, content: str) -> None: ...

Methods

emit()

Emit a raw stream event to the host. Parameters:

Parameter	Type	Required	Description
`event_type`	`str`	Yes	Event type identifier
`data`	`dict \| str`	Yes	Event payload (dict serialized to JSON, or string)

Example:

ctx.stream.emit("custom-phase", {"step": 3, "total": 10})
ctx.stream.emit("debug", "Processing complete")

progress()

Emit a data-tool-progress event for UI progress display. Parameters:

Parameter	Type	Required	Default	Description
`content`	`str`	Yes	-	Progress message
`tool_name`	`str \| None`	No	`None`	Tool identifier for grouping

Example:

ctx.stream.progress("Starting analysis...")
ctx.stream.progress("Fetching repository data", tool_name="GitHub")
ctx.stream.progress("Analysing code patterns", tool_name="Analyser")

intent()

Emit a data-intent event for high-level state changes. Parameters:

Parameter	Type	Required	Description
`content`	`str`	Yes	Intent description

Example:

ctx.stream.intent("Discovering repository structure")
ctx.stream.intent("Identifying security issues")
ctx.stream.intent("Generating recommendations")

Common Usage Patterns

Phase-Based Progress

def execute(prompt, ctx):
    ctx.stream.progress("Phase 1: Parsing input")
    config = parse_input(prompt)

    ctx.stream.progress("Phase 2: Fetching data", tool_name="GitHub")
    data = ctx.tools.call("fetch_repo", config)

    ctx.stream.progress("Phase 3: Analysing", tool_name="LLM")
    analysis = ctx.llm.generate(...)

    ctx.stream.progress("Phase 4: Finalising")
    return ok({"result": analysis.text})

Intent for State Changes

def execute(prompt, ctx):
    ctx.stream.intent("Understanding task requirements")
    requirements = extract_requirements(prompt)

    ctx.stream.intent("Planning approach")
    plan = create_plan(requirements)

    ctx.stream.intent("Executing plan")
    for step in plan.steps:
        ctx.stream.progress(f"Step {step.number}: {step.description}")
        execute_step(step)

    ctx.stream.intent("Finalising results")
    return ok({"completed": True})

Tool-Associated Progress

def execute(prompt, ctx):
    ctx.stream.progress("Initialising", tool_name="Setup")

    ctx.stream.progress("Querying database", tool_name="PostgreSQL")
    rows = ctx.tools.call("query", {"sql": "SELECT ..."})

    ctx.stream.progress("Processing results", tool_name="Processor")
    processed = [transform(r) for r in rows]

    ctx.stream.progress("Storing analysis", tool_name="Storage")
    ctx.http.fetch(..., method="POST", body=json.dumps(processed))

    ctx.stream.progress("Complete", tool_name="Setup")
    return ok({"count": len(processed)})

Fallback When Unavailable

ctx.stream is always present. It is a safe no-op in test contexts:

def execute(prompt, ctx):
    # Safe wrapper
    def progress(msg, tool=None):
        if ctx.stream:
            ctx.stream.progress(msg, tool_name=tool)

    progress("Starting...")

    # Work...

    progress("Complete")
    return ok({"done": True})

Emission During LLM Calls

Progress emits are fire-and-forget over NATS - they do not block:

ctx.stream.progress("Starting LLM call...")  # Sent immediately

# LLM call blocks until response; progress already sent
result = ctx.llm.generate(messages, model="claude-sonnet-4-6")

# Back in Python
ctx.stream.progress("LLM complete")  # Sent now

The host may emit its own progress events during the suspension.

Event Types

Standard types used by Friday:

Type	Usage
`data-tool-progress`	Agent progress updates (use `progress()`)
`data-intent`	High-level state changes (use `intent()`)
`data-error`	Error events (usually emitted by host)

Custom types can be emitted via emit() but may not have UI handlers.

Best Practices

Emit before expensive operations - Warn users before long LLM calls
Use tool_name for grouping - Helps UI organise progress by component
Keep messages concise - 50-100 characters ideal for UI display
Avoid tight loop emission - Batch or debounce high-frequency updates
Prefer intent for phases, progress for detail - Two-level hierarchy
Always safe to call - ctx.stream never None, but may no-op in tests

When to Emit

Scenario	Method	Example
Starting a phase	`intent()`	”Analysing repository”
Detailed progress	`progress()`	”Fetching 50 files…”
Tool-specific work	`progress(tool_name=...)`	tool_name=“GitHub”
Fallback scenarios	`progress()`	”Retrying with alternate model…”
Completion	`intent()`	”Analysis complete”

How to Stream Progress

Task-oriented guide

How Agents Work

The subprocess model and host capabilities

Getting started

Core concepts

Guides

Agent SDK

Reference

Resources

Class: StreamEmitter

Methods

emit()

progress()

intent()

Common Usage Patterns

Phase-Based Progress

Intent for State Changes

Tool-Associated Progress

Fallback When Unavailable

Emission During LLM Calls

Event Types

Best Practices

When to Emit

See Also

How to Stream Progress

How Agents Work

​Class: StreamEmitter

​Methods

​emit()

​progress()

​intent()

​Common Usage Patterns

​Phase-Based Progress

​Intent for State Changes

​Tool-Associated Progress

​Fallback When Unavailable

​Emission During LLM Calls

​Event Types

​Best Practices

​When to Emit

​See Also

How to Stream Progress

How Agents Work

Class: StreamEmitter

Methods

emit()

progress()

intent()

Common Usage Patterns

Phase-Based Progress

Intent for State Changes

Tool-Associated Progress

Fallback When Unavailable

Emission During LLM Calls

Event Types

Best Practices

When to Emit

See Also