Live progress and cancellation for page sync

Agno's Knowledge.stream_sync_pages() yields a PageSyncProgress snapshot as a sync moves through waiting, discovered, publishing and pruning, then ends with the usual SyncReport. Each snapshot carries absolute counts and the path of the page it just processed, so a slow consumer that skips a few still reads the right totals.

We ran it against llmstxt.org with a local FastEmbed embedder, and the real stream looks like this.

from agno.knowledge.page import PageSyncProgress
 
knowledge.setup()
for event in knowledge.stream_sync_pages(url="https://llmstxt.org/llms.txt"):
    if isinstance(event, PageSyncProgress):
        print(
            f"{event.stage:<11} {event.processed}/{event.discovered} "
            f"updated={event.updated} failed={event.failed} {event.path or ''}"
        )
    else:
        print(repr(event))

waiting     0/0 updated=0 failed=0
discovered  0/3 updated=0 failed=0
publishing  1/3 updated=1 failed=0 /index.md
publishing  2/3 updated=2 failed=0 /intro.html.md
publishing  3/3 updated=3 failed=0 /ed.md
pruning     3/3 updated=3 failed=0
SyncReport(schema_version=1, status='completed', discovered=3, updated=3, deleted=0, failed=0, unknown=0, errors=(), failed_paths=(), skipped=0, skipped_paths=())

astream_sync_pages() does the same in async code. If you keep calling sync_pages or async_sync_pages, pass on_progress= a short callback and it receives the same snapshots.

To show that progress in an app, run the sync inside a workflow. A function step can now yield StepProgress(content=..., data=...) before its StepOutput, and a streaming run turns each one into a StepProgressEvent under the same run ID, step and retry attempt. AgentOS sends it over the existing workflow SSE route, and AgentOSClient.run_workflow_stream() parses it. The cookbook's sync step, trimmed, does exactly that.

from agno.knowledge.page import SyncReport
from agno.workflow import Step, Workflow
from agno.workflow.types import StepInput, StepOutput, StepProgress
 
 
async def sync_source(step_input: StepInput):
    updates = knowledge.astream_sync_pages(url="https://llmstxt.org/llms.txt")
    try:
        async for update in updates:
            if isinstance(update, SyncReport):
                yield StepOutput(content=update.model_dump(), success=update.status != "partial")
            else:
                yield StepProgress(
                    content=f"Processed {update.processed} of {update.discovered} pages",
                    data=update.model_dump(),
                )
    finally:
        # Closing the stream cancels the sync worker when the step is stopped early.
        await updates.aclose()
 
 
sync = Workflow(id="sync-docs", db=db, steps=[Step(name="reconcile", executor=sync_source)])

Cancelling that run now stops the work. Earlier releases let a cancelled function step run to its end. The step checks the workflow run between the items the function yields, closes the function's generator so its finally runs, and raises RunCancelledException. Closing the Knowledge stream cancels the page worker. It stops at its next cancellation check and releases the sync lock. Cancellation is cooperative, so a function that never yields can't be stopped. A tiny three-page step with no model, cancelled at its first progress event, printed the lines below.

WorkflowStarted
StepStarted
StepProgress index 1 Processed 1 of 3 pages {'path': '/index.md'}
cleanup ran
WorkflowCancelled
WorkflowCompleted

AgentOS saves every workflow event, and one progress event per page adds up on a large site. The cookbook sets events_to_skip=[WorkflowRunEvent.step_progress] so progress streams live without filling the run's stored history.

See the cookbook, and learn more about published pages and workflow cancellation in the documentation.