agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: Add long-running agents and background responses support (#3808)

Browse Source 
* Python: Add long-running agents and background responses support

- Add ContinuationToken TypedDict to core types
- Add continuation_token field to ChatResponse, ChatResponseUpdate,
  AgentResponse, and AgentResponseUpdate
- Add background and continuation_token options to OpenAIResponsesOptions
- Implement polling via responses.retrieve() and streaming resumption
  in RawOpenAIResponsesClient
- Propagate continuation tokens through agent run() and
  map_chat_to_agent_update
- Fix streaming telemetry 'Failed to detach context' error in both
  ChatTelemetryLayer and AgentTelemetryLayer by avoiding
  trace.use_span() context attachment for async-managed spans
- Add 14 unit tests for continuation token types and background flows
- Add background_responses sample showing polling and stream resumption

Fixes #2478

* Python: Add A2A long-running task support via ContinuationToken

- Make ContinuationToken provider-agnostic (total=False, optional task_id/context_id fields)
- Add background param to A2AAgent.run() controlling token emission
- Add poll_task() for single-request task state retrieval
- Add resubscribe support via continuation_token param on run()
- Extract _updates_from_task() and _map_a2a_stream() for cleaner code
- Streamline run()/streaming by removing intermediate _stream_updates wrapper
- Update A2A sample to show background=False (default) with link to background_responses sample
- Remove stale BareAgent from __all__
- Add 12 new A2A continuation token tests

* fix logic for overriding continuation token when done

* refactored ContinuationToken setup
This commit is contained in:
Eduard van Valkenburg
2026-02-10 21:37:43 +01:00
committed by GitHub
Unverified
parent 32ba81e990
commit 35097d8c75
12 changed files with 916 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/a2a/agent_framework_a2a/__init__.py
+2 -1
View File
@@ -2,7 +2,7 @@
import importlib.metadata
from ._agent import A2AAgent
from ._agent import A2AAgent, A2AContinuationToken
try:
__version__ = importlib.metadata.version(__name__)
@@ -11,5 +11,6 @@ except importlib.metadata.PackageNotFoundError:
__all__ = [
"A2AAgent",
"A2AContinuationToken",
"__version__",
]
python/packages/a2a/agent_framework_a2a/_agent.py
+126 -76
View File
@@ -20,6 +20,8 @@ from a2a.types import (
FileWithUri,
Message,
Task,
TaskIdParams,
TaskQueryParams,
TaskState,
TextPart,
TransportProtocol,
@@ -34,21 +36,39 @@ from agent_framework import (
BaseAgent,
ChatMessage,
Content,
ContinuationToken,
ResponseStream,
normalize_messages,
prepend_agent_framework_to_user_agent,
)
from agent_framework.observability import AgentTelemetryLayer
__all__ = ["A2AAgent"]
__all__ = ["A2AAgent", "A2AContinuationToken"]
URI_PATTERN = re.compile(r"^data:(?P<media_type>[^;]+);base64,(?P<base64_data>[A-Za-z0-9+/=]+)$")
class A2AContinuationToken(ContinuationToken):
"""Continuation token for A2A protocol long-running tasks."""
task_id: str
"""A2A protocol task ID."""
context_id: str
"""A2A protocol context ID."""
TERMINAL_TASK_STATES = [
TaskState.completed,
TaskState.failed,
TaskState.canceled,
TaskState.rejected,
]
IN_PROGRESS_TASK_STATES = [
TaskState.submitted,
TaskState.working,
TaskState.input_required,
TaskState.auth_required,
]
def _get_uri_data(uri: str) -> str:
@@ -193,6 +213,8 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
*,
stream: Literal[False] = ...,
thread: AgentThread | None = None,
continuation_token: A2AContinuationToken | None = None,
background: bool = False,
**kwargs: Any,
) -> Awaitable[AgentResponse[Any]]: ...
@@ -203,6 +225,8 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
*,
stream: Literal[True],
thread: AgentThread | None = None,
continuation_token: A2AContinuationToken | None = None,
background: bool = False,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
@@ -212,85 +236,62 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
*,
stream: bool = False,
thread: AgentThread | None = None,
continuation_token: A2AContinuationToken | None = None,
background: bool = False,
**kwargs: Any,
) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
"""Get a response from the agent.
This method returns the final result of the agent's execution
as a single AgentResponse object when stream=False. When stream=True,
it returns a ResponseStream that yields AgentResponseUpdate objects.
Args:
messages: The message(s) to send to the agent.
Keyword Args:
stream: Whether to stream the response. Defaults to False.
thread: The conversation thread associated with the message(s).
continuation_token: Optional token to resume a long-running task
instead of starting a new one.
background: When True, in-progress task updates surface continuation
tokens so the caller can poll or resubscribe later. When False
(default), the agent internally waits for the task to complete.
kwargs: Additional keyword arguments.
Returns:
When stream=False: An Awaitable[AgentResponse].
When stream=True: A ResponseStream of AgentResponseUpdate items.
"""
if continuation_token is not None:
a2a_stream: AsyncIterable[Any] = self.client.resubscribe(TaskIdParams(id=continuation_token["task_id"]))
else:
normalized_messages = normalize_messages(messages)
a2a_message = self._prepare_message_for_a2a(normalized_messages[-1])
a2a_stream = self.client.send_message(a2a_message)
response = ResponseStream(
self._map_a2a_stream(a2a_stream, background=background),
finalizer=lambda updates: AgentResponse.from_updates(list(updates)),
)
if stream:
return self._run_stream_impl(messages=messages, thread=thread, **kwargs)
return self._run_impl(messages=messages, thread=thread, **kwargs)
return response
return response.get_final_response()
async def _run_impl(
async def _map_a2a_stream(
self,
messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
a2a_stream: AsyncIterable[Any],
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AgentResponse[Any]:
"""Non-streaming implementation of run."""
# Collect all updates and use framework to consolidate updates into response
updates: list[AgentResponseUpdate] = []
async for update in self._stream_updates(messages, thread=thread, **kwargs):
updates.append(update)
return AgentResponse.from_updates(updates)
def _run_stream_impl(
self,
messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
"""Streaming implementation of run."""
def _finalize(updates: Sequence[AgentResponseUpdate]) -> AgentResponse[Any]:
return AgentResponse.from_updates(list(updates))
return ResponseStream(self._stream_updates(messages, thread=thread, **kwargs), finalizer=_finalize)
async def _stream_updates(
self,
messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
background: bool = False,
) -> AsyncIterable[AgentResponseUpdate]:
"""Internal method to stream updates from the A2A agent.
"""Map raw A2A protocol items to AgentResponseUpdates.
Args:
messages: The message(s) to send to the agent.
a2a_stream: The raw A2A event stream.
Keyword Args:
thread: The conversation thread associated with the message(s).
kwargs: Additional keyword arguments.
Yields:
AgentResponseUpdate items from the A2A agent.
background: When False, in-progress task updates are silently
consumed (the stream keeps iterating until a terminal state).
When True, they are yielded with a continuation token.
"""
normalized_messages = normalize_messages(messages)
a2a_message = self._prepare_message_for_a2a(normalized_messages[-1])
response_stream = self.client.send_message(a2a_message)
async for item in response_stream:
async for item in a2a_stream:
if isinstance(item, Message):
# Process A2A Message
contents = self._parse_contents_from_a2a(item.parts)
yield AgentResponseUpdate(
contents=contents,
@@ -300,33 +301,82 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
)
elif isinstance(item, tuple) and len(item) == 2: # ClientEvent = (Task, UpdateEvent)
task, _update_event = item
if isinstance(task, Task) and task.status.state in TERMINAL_TASK_STATES:
# Convert Task artifacts to ChatMessages and yield as separate updates
task_messages = self._parse_messages_from_task(task)
if task_messages:
for message in task_messages:
# Use the artifact's ID from raw_representation as message_id for unique identification
artifact_id = getattr(message.raw_representation, "artifact_id", None)
yield AgentResponseUpdate(
contents=message.contents,
role=message.role,
response_id=task.id,
message_id=artifact_id,
raw_representation=task,
)
else:
# Empty task
yield AgentResponseUpdate(
contents=[],
role="assistant",
response_id=task.id,
raw_representation=task,
)
if isinstance(task, Task):
for update in self._updates_from_task(task, background=background):
yield update
else:
# Unknown response type
msg = f"Only Message and Task responses are supported from A2A agents. Received: {type(item)}"
raise NotImplementedError(msg)
# ------------------------------------------------------------------
# Task helpers
# ------------------------------------------------------------------
def _updates_from_task(self, task: Task, *, background: bool = False) -> list[AgentResponseUpdate]:
"""Convert an A2A Task into AgentResponseUpdate(s).
Terminal tasks produce updates from their artifacts/history.
In-progress tasks produce a continuation token update only when
``background=True``; otherwise they are silently skipped so the
caller keeps consuming the stream until completion.
"""
if task.status.state in TERMINAL_TASK_STATES:
task_messages = self._parse_messages_from_task(task)
if task_messages:
return [
AgentResponseUpdate(
contents=message.contents,
role=message.role,
response_id=task.id,
message_id=getattr(message.raw_representation, "artifact_id", None),
raw_representation=task,
)
for message in task_messages
]
return [AgentResponseUpdate(contents=[], role="assistant", response_id=task.id, raw_representation=task)]
if background and task.status.state in IN_PROGRESS_TASK_STATES:
token = self._build_continuation_token(task)
return [
AgentResponseUpdate(
contents=[],
role="assistant",
response_id=task.id,
continuation_token=token,
raw_representation=task,
)
]
return []
@staticmethod
def _build_continuation_token(task: Task) -> A2AContinuationToken | None:
"""Build an A2AContinuationToken from an A2A Task if it is still in progress."""
if task.status.state in IN_PROGRESS_TASK_STATES:
return A2AContinuationToken(task_id=task.id, context_id=task.context_id)
return None
async def poll_task(self, continuation_token: A2AContinuationToken) -> AgentResponse[Any]:
"""Poll for the current state of a long-running A2A task.
Unlike ``run(continuation_token=...)``, which resubscribes to the SSE
stream, this performs a single request to retrieve the task state.
Args:
continuation_token: A token previously obtained from a response's
``continuation_token`` field.
Returns:
An AgentResponse whose ``continuation_token`` is set when the task
is still in progress, or ``None`` when it has reached a terminal state.
"""
task_id = continuation_token["task_id"]
task = await self.client.get_task(TaskQueryParams(id=task_id))
updates = self._updates_from_task(task, background=True)
if updates:
return AgentResponse.from_updates(updates)
return AgentResponse(messages=[], response_id=task.id, raw_representation=task)
def _prepare_message_for_a2a(self, message: ChatMessage) -> A2AMessage:
"""Prepare a ChatMessage for the A2A protocol.
python/packages/a2a/tests/test_a2a_agent.py
+186
View File
@@ -29,6 +29,7 @@ from agent_framework import (
from agent_framework.a2a import A2AAgent
from pytest import fixture, raises
from agent_framework_a2a import A2AContinuationToken
from agent_framework_a2a._agent import _get_uri_data # type: ignore
@@ -38,6 +39,8 @@ class MockA2AClient:
def __init__(self) -> None:
self.call_count: int = 0
self.responses: list[Any] = []
self.resubscribe_responses: list[Any] = []
self.get_task_response: Task | None = None
def add_message_response(self, message_id: str, text: str, role: str = "agent") -> None:
"""Add a mock Message response."""
@@ -80,6 +83,18 @@ class MockA2AClient:
client_event = (task, update_event)
self.responses.append(client_event)
def add_in_progress_task_response(
self,
task_id: str,
context_id: str = "test-context",
state: TaskState = TaskState.working,
) -> None:
"""Add a mock in-progress Task response (non-terminal)."""
status = TaskStatus(state=state, message=None)
task = Task(id=task_id, context_id=context_id, status=status)
client_event = (task, None)
self.responses.append(client_event)
async def send_message(self, message: Any) -> AsyncIterator[Any]:
"""Mock send_message method that yields responses."""
self.call_count += 1
@@ -88,6 +103,22 @@ class MockA2AClient:
response = self.responses.pop(0)
yield response
async def resubscribe(self, request: Any) -> AsyncIterator[Any]:
"""Mock resubscribe method that yields responses."""
self.call_count += 1
for response in self.resubscribe_responses:
yield response
self.resubscribe_responses.clear()
async def get_task(self, request: Any) -> Task:
"""Mock get_task method that returns a task."""
self.call_count += 1
if self.get_task_response is not None:
return self.get_task_response
msg = "No get_task response configured"
raise ValueError(msg)
@fixture
def mock_a2a_client() -> MockA2AClient:
@@ -598,3 +629,158 @@ def test_a2a_agent_initialization_with_timeout_parameter() -> None:
# Verify it's an httpx.Timeout object with our custom timeout applied to all components
assert isinstance(timeout_arg, httpx.Timeout)
# region Continuation Token Tests
async def test_working_task_emits_continuation_token(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
"""Test that a working (non-terminal) task yields an update with a continuation token when background=True."""
mock_a2a_client.add_in_progress_task_response("task-wip", context_id="ctx-1", state=TaskState.working)
response = await a2a_agent.run("Start long task", background=True)
assert isinstance(response, AgentResponse)
assert response.continuation_token is not None
assert response.continuation_token["task_id"] == "task-wip"
assert response.continuation_token["context_id"] == "ctx-1"
async def test_submitted_task_emits_continuation_token(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
"""Test that a submitted task yields a continuation token when background=True."""
mock_a2a_client.add_in_progress_task_response("task-sub", state=TaskState.submitted)
response = await a2a_agent.run("Submit task", background=True)
assert response.continuation_token is not None
assert response.continuation_token["task_id"] == "task-sub"
async def test_input_required_task_emits_continuation_token(
a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient
) -> None:
"""Test that an input_required task yields a continuation token when background=True."""
mock_a2a_client.add_in_progress_task_response("task-input", state=TaskState.input_required)
response = await a2a_agent.run("Need input", background=True)
assert response.continuation_token is not None
assert response.continuation_token["task_id"] == "task-input"
async def test_working_task_no_token_without_background(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
"""Test that background=False (default) does not emit continuation tokens for in-progress tasks."""
mock_a2a_client.add_in_progress_task_response("task-fg", context_id="ctx-fg", state=TaskState.working)
response = await a2a_agent.run("Foreground task")
assert response.continuation_token is None
async def test_completed_task_has_no_continuation_token(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
"""Test that a completed task does not set a continuation token."""
mock_a2a_client.add_task_response("task-done", [{"id": "art-1", "content": "Result"}])
response = await a2a_agent.run("Quick task")
assert response.continuation_token is None
assert len(response.messages) == 1
assert response.messages[0].text == "Result"
async def test_streaming_emits_continuation_token(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
"""Test that streaming with background=True yields updates with continuation tokens."""
mock_a2a_client.add_in_progress_task_response("task-stream", context_id="ctx-s", state=TaskState.working)
updates: list[AgentResponseUpdate] = []
async for update in a2a_agent.run("Stream task", stream=True, background=True):
updates.append(update)
assert len(updates) == 1
assert updates[0].continuation_token is not None
assert updates[0].continuation_token["task_id"] == "task-stream"
assert updates[0].continuation_token["context_id"] == "ctx-s"
async def test_resume_via_continuation_token(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
"""Test that run() with continuation_token uses resubscribe instead of send_message."""
# Set up the resubscribe response (completed task)
status = TaskStatus(state=TaskState.completed, message=None)
artifact = Artifact(
artifact_id="art-resume",
name="result",
parts=[Part(root=TextPart(text="Resumed result"))],
)
task = Task(id="task-resume", context_id="ctx-r", status=status, artifacts=[artifact])
mock_a2a_client.resubscribe_responses.append((task, None))
token = A2AContinuationToken(task_id="task-resume", context_id="ctx-r")
response = await a2a_agent.run(continuation_token=token)
assert isinstance(response, AgentResponse)
assert len(response.messages) == 1
assert response.messages[0].text == "Resumed result"
assert response.continuation_token is None
async def test_resume_streaming_via_continuation_token(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
"""Test that streaming run() with continuation_token and background=True uses resubscribe."""
# Still working
status_wip = TaskStatus(state=TaskState.working, message=None)
task_wip = Task(id="task-rs", context_id="ctx-rs", status=status_wip)
# Then completed
status_done = TaskStatus(state=TaskState.completed, message=None)
artifact = Artifact(
artifact_id="art-rs",
name="result",
parts=[Part(root=TextPart(text="Stream resumed"))],
)
task_done = Task(id="task-rs", context_id="ctx-rs", status=status_done, artifacts=[artifact])
mock_a2a_client.resubscribe_responses.extend([(task_wip, None), (task_done, None)])
token = A2AContinuationToken(task_id="task-rs", context_id="ctx-rs")
updates: list[AgentResponseUpdate] = []
async for update in a2a_agent.run(stream=True, continuation_token=token, background=True):
updates.append(update)
# First update: in-progress with token, second: completed with content
assert len(updates) == 2
assert updates[0].continuation_token is not None
assert updates[0].continuation_token["task_id"] == "task-rs"
assert updates[1].continuation_token is None
assert updates[1].contents[0].text == "Stream resumed"
async def test_poll_task_in_progress(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
"""Test poll_task returns continuation token when task is still in progress."""
status = TaskStatus(state=TaskState.working, message=None)
mock_a2a_client.get_task_response = Task(id="task-poll", context_id="ctx-p", status=status)
token = A2AContinuationToken(task_id="task-poll", context_id="ctx-p")
response = await a2a_agent.poll_task(token)
assert response.continuation_token is not None
assert response.continuation_token["task_id"] == "task-poll"
async def test_poll_task_completed(a2a_agent: A2AAgent, mock_a2a_client: MockA2AClient) -> None:
"""Test poll_task returns result with no continuation token when task is complete."""
status = TaskStatus(state=TaskState.completed, message=None)
artifact = Artifact(
artifact_id="art-poll",
name="result",
parts=[Part(root=TextPart(text="Poll result"))],
)
mock_a2a_client.get_task_response = Task(
id="task-poll-done", context_id="ctx-pd", status=status, artifacts=[artifact]
)
token = A2AContinuationToken(task_id="task-poll-done", context_id="ctx-pd")
response = await a2a_agent.poll_task(token)
assert response.continuation_token is None
assert len(response.messages) == 1
assert response.messages[0].text == "Poll result"
# endregion