Python: [Feature]: First-class background execution for workflows (parity with agent Background Responses)

[likebean]

Description

Agents already have Background Responses: start a long run, get a continuation_token, poll or resume until the token is None.

response = await agent.run(
    messages="...",
    session=session,
    options={"background": True},
)
while response.continuation_token is not None:
    await asyncio.sleep(2)
    response = await agent.run(
        session=session,
        options={"continuation_token": response.continuation_token},
    )

We want the same submit → handle → poll to a terminal state experience for workflows, including when the workflow is exposed as an agent.

What we tried / observed:

1. workflow.run(..., options={"background": True}) is not a supported API.
2. workflow.as_agent().run(..., options={"background": True}) also does not work. WorkflowAgent.run() does not accept options, and internally calls workflow.run() (checkpoint / responses / checkpoint_id), so the OpenAI/Azure Responses background + continuation_token path never applies.
3. Core workflows can persist and resume via checkpoint storage (checkpoint_id + checkpoint_storage). That is pause/resume of the graph, not a first-class background run handle.
4. Foundry hosting can do protocol-level background=true for workflow agents when ResponsesServerOptions(resilient_background=True) is set. That is host/protocol recovery, not an in-process Workflow.run / WorkflowAgent.run option. Regular (non-workflow) agents cannot use resilient_background.

Related but broader: #1441 (timeouts, durable timers, executor polling). This issue is specifically about API parity with agent Background Responses for a whole workflow run.

What we are asking

1. What is the recommended way today to start a workflow in the background and poll for completion / resume after disconnect or process restart?
2. Should platforms treat this as:
   • wrap workflow.run(...) in our own task + persist checkpoint_id, or
   • host the workflow as an agent and use Foundry/Responses background=true, or
   • something else?
3. Is a first-class API planned, for example:

result = await workflow.run(
    message="...",
    checkpoint_storage=storage,
    options={"background": True},
)
# result.continuation_token or result.run_id for polling

or the same shape on workflow.as_agent().run(...)?

Expected behavior we want: one stable handle for the workflow run, pollable status (running / completed / failed / paused), and resume after client disconnect without inventing a second job system.

Code Sample

from agent_framework import InMemoryCheckpointStorage

workflow = build_workflow()  # WorkflowBuilder(...).build()

# Desired (does not exist today)
agent = workflow.as_agent(name="report-workflow")
response = await agent.run(
    "Generate the weekly report",
    options={"background": True},
)
while response.continuation_token is not None:
    response = await agent.run(options={"continuation_token": response.continuation_token})

What actually happens: TypeError: WorkflowAgent.run() got an unexpected keyword argument 'options'.

Checkpoint-based alternative we know about:

storage = InMemoryCheckpointStorage()
# run until interrupted, then:
latest = await storage.get_latest(workflow_name=workflow.name)
await workflow.run(checkpoint_id=latest.checkpoint_id, checkpoint_storage=storage)

That works for resume, but it is not the same as submitting a background run and polling a framework-issued handle.

Language/SDK

Both (question is about the shared model; we are using Python)

Additional context

Docs that led us here:

• https://learn.microsoft.com/en-us/agent-framework/agents/background-responses
• python/samples/02-agents/background_responses.py
• python/samples/03-workflows/checkpoint/checkpoint_with_resume.py
• python/samples/03-workflows/checkpoint/workflow_as_agent_checkpoint.py
• python/samples/04-hosting/foundry-hosted-agents/responses/resilient_long_running_workflow/

[aashikantkumar]

Hi Eduard van Valkenburg (@eavanvalkenburg),

I’m interested in contributing to the Python side of this issue. Before starting an implementation, I reviewed the current execution paths through Agent.run(), WorkflowAgent.run(), Workflow.run(), workflow checkpointing, Foundry resilient-background hosting, and the Durable Extension.

My current understanding is that adding an options parameter to WorkflowAgent.run() would remove the immediate TypeError, but it would not be sufficient to provide background execution for the entire workflow.

For a regular OpenAI/Azure Responses agent, background=True is a provider-level option: the remote service owns the operation and returns a provider-specific continuation token. A workflow, however, is provider-independent, may contain multiple agents and ordinary Python executors, and normally executes inside the application process.

Therefore, whole-workflow background execution seems to be a workflow run-lifecycle and hosting concern rather than a chat-client option.

Possible design direction

One possible direction may be to introduce a small, provider-neutral workflow run abstraction in core, with the exact API shape and naming open for discussion.

For example:

run = await workflow.submit(
    message="Generate the weekly report",
    backend=run_backend,
)

snapshot = await run_backend.get_run(run.run_id)

if snapshot.status == WorkflowRunStatus.SUCCEEDED:
    result = await run_backend.get_result(run.run_id)

The following identities could remain separate:

• session_id: identifies a conversation or session.
• run_id: identifies one submitted workflow execution.
• checkpoint_id: identifies a saved boundary within that execution.
• response_id: identifies an agent or protocol response.

A possible backend protocol could expose operations such as:

class WorkflowRunBackend(Protocol):
    async def submit(...) -> WorkflowRun: ...
    async def get_run(self, run_id: str) -> WorkflowRun: ...
    async def get_result(self, run_id: str) -> WorkflowRunResult: ...
    async def resume(...) -> WorkflowRun: ...
    async def cancel(self, run_id: str) -> WorkflowRun: ...

Potential implementations could provide different execution guarantees:

1. A local backend using asyncio.Task, clearly documented as not surviving process termination.
2. A Durable Task backend supplied by the Durable Extension for persistent and distributed execution.
3. A Foundry/Responses backend mapping the workflow lifecycle to hosted response IDs and resilient-background execution.

This could provide a consistent:

submit → run_id → poll/resume → terminal result

experience without requiring agent-framework-core to implement another durable queue, database, worker-leasing system, and crash-recovery runtime.

I also think a workflow-level run_id may be clearer than reusing continuation_token. Continuation tokens are currently provider-specific and opaque, while a workflow may contain multiple providers and non-agent executors.

Proposed implementation sequence

If this direction matches the project’s intended architecture, I would be happy to begin with a focused initial PR:

1. Agree on lifecycle semantics and public API names.
2. Add the workflow run models and backend protocol.
3. Add a clearly non-durable local backend.
4. Add deterministic offline tests for submission, polling, completion, failure, and cancellation.
5. Document the local backend’s durability limitations.

I do not currently have access to paid OpenAI or Foundry credentials. However, the provider-neutral lifecycle behavior can be tested using deterministic executors, fake agents, and in-memory or file-backed storage. Provider-specific end-to-end validation could remain in the corresponding integration test suites.

Before I begin implementation, could you please confirm:

1. Should whole-workflow background execution be represented through a workflow runtime API such as submit(), instead of chat-client options?
2. Would a pluggable WorkflowRunBackend-style boundary fit the intended architecture?
3. Should the Durable Extension be considered the recommended solution for process-restart recovery today?
4. Is this issue open for a community implementation contribution?
5. If so, what would be the most useful scope for the first PR?

I’m happy to adjust the proposal to match the team’s direction before starting implementation.

Thank you!