[BUG] A2A (a2a-sdk 1.x): follow-up messages on an existing task run the executor with the first request's contextvars #690

Summary

With build_a2a_app on a2a-sdk 1.x, AgentExecutor.execute() for a follow-up message on an existing task (for example, a reply after the task went input-required) runs with the contextvars of the request that created the task, not those of the request that carried the follow-up. This affects:

• BedrockAgentCoreContext: the workload access token, request id, session id, OAuth2 callback URL and forwarded request headers (including Authorization) all return the first request's values on every later turn.

• AgentCore Identity uses the stale values. @requires_access_token, @requires_api_key and @requires_wat read the workload access token from BedrockAgentCoreContext (identity/auth.py). So does token propagation on InvokeAgentRuntime and InvokeGateway (_utils/identity_propagation.py). While handling turn 2 they all use turn 1's token and callback URL.

• Any per-request contextvar that ASGI middleware sets on the app build_a2a_app returns (authentication, tenant or caller identity, request-scoped tracing) is stale too. The executor sees turn 1's value on every turn of the task.

ServerCallContext.state is correct per message; only contextvars are stale. Both message/send and message/stream are affected, through build_a2a_app and serve_a2a. The cause is not timing: turn 2 sent immediately, after a delay, or while turn 1 is still executing all see turn 1's values.

Scope. Only follow-up messages on the same task id handled by the same process are affected. That includes any task parked in input-required or auth-required, whose ActiveTask is not cleaned up (a2a-python [#1296](a2aproject/a2a-python#1296)). These are not affected:

• a new task per turn under the same context_id

• a follow-up after a terminal state, which is rejected

• a follow-up served by a different process

Environment

• bedrock-agentcore 1.24.0, which matches main (build_a2a_app and serve_a2a, default BedrockCallContextBuilder)

• a2a-sdk 1.0.3 and 1.2.1 (DefaultRequestHandler, which is DefaultRequestHandlerV2)

• CPython 3.12, on a uvicorn server and in-process

a2a-sdk 0.3.x is not affected: its request handler starts a new producer per message (default_request_handler.py, 0.3.7).

Cause

In a2a-sdk 1.x, DefaultRequestHandler keeps one ActiveTask per task id (ActiveTaskRegistry.get_or_create). ActiveTask.start() creates the producer with asyncio.create_task(self._run_producer()) during the first request for the task, so the producer copies that request's contextvars once. Later requests on the same task call ActiveTask.enqueue_request(), which only puts the RequestContext on a queue, and the long-lived producer calls executor.execute() from its own (first-request) context.

BedrockCallContextBuilder.build() sets BedrockAgentCoreContext correctly in each request's context, but that context never reaches the executor after the first message on a task. The window can be long: a task parked in input-required keeps its ActiveTask (see a2a-python [#1296](a2aproject/a2a-python#1296)), so a reply minutes or hours later still runs in the original request's context, and a short-lived token held there may have expired.

Reproduction

Self-contained; runs in-process with httpx.ASGITransport, no AWS calls. Each HTTP request runs in a fresh contextvars.Context, as it would on a real server.

import asyncio, contextvars, uuid

import httpx
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.types import Message, Part, Role, SendMessageRequest, Task, TaskState, TaskStatus, TaskStatusUpdateEvent
from bedrock_agentcore.runtime import BedrockAgentCoreContext, build_a2a_app

REQUEST_TAG: contextvars.ContextVar[str] = contextvars.ContextVar("request_tag", default="unset")


class TagMiddleware:  # stands in for any per-request middleware (auth, tenant, tracing)
    def __init__(self, app): self.app = app
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            token = REQUEST_TAG.set(dict(scope["headers"]).get(b"x-request-tag", b"unset").decode())
            try:
                return await self.app(scope, receive, send)
            finally:
                REQUEST_TAG.reset(token)
        return await self.app(scope, receive, send)


seen = []


class TwoTurnExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        seen.append((context.get_user_input(), REQUEST_TAG.get(), BedrockAgentCoreContext.get_workload_access_token()))
        first = context.current_task is None
        if first:
            await event_queue.enqueue_event(Task(id=context.task_id, context_id=context.context_id,
                                                 status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED)))
        await event_queue.enqueue_event(TaskStatusUpdateEvent(
            task_id=context.task_id, context_id=context.context_id,
            status=TaskStatus(state=TaskState.TASK_STATE_INPUT_REQUIRED if first else TaskState.TASK_STATE_COMPLETED)))

    async def cancel(self, context, event_queue): return None


app = build_a2a_app(TwoTurnExecutor(), runtime_url="http://t/")
app.add_middleware(TagMiddleware)


async def send(tag, text, ids=None):
    http = httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://t", timeout=30,
                             headers={"x-request-tag": tag, "WorkloadAccessToken": f"WAT-{tag}"})
    card = await A2ACardResolver(http, "http://t").get_agent_card()
    client = ClientFactory(ClientConfig(httpx_client=http)).create(card)
    msg = Message(role=Role.ROLE_USER, message_id=str(uuid.uuid4()), parts=[Part(text=text)])
    if ids:
        msg.task_id, msg.context_id = ids
    last = None
    async for ev in client.send_message(SendMessageRequest(message=msg)):
        last = ev
    await http.aclose()
    return last


async def fresh(coro):
    return await asyncio.get_running_loop().create_task(coro, context=contextvars.Context())


async def main():
    r1 = await fresh(send("request-1", "book a flight"))
    ids = (r1.task.id, r1.task.context_id) if r1.HasField("task") else (r1.status_update.task_id, r1.status_update.context_id)
    await fresh(send("request-2", "Friday", ids))
    for row in seen:
        print("execute() saw:", row)


asyncio.run(main())

Actual:

execute() saw: ('book a flight', 'request-1', 'WAT-request-1')
execute() saw: ('Friday', 'request-1', 'WAT-request-1')

Expected:

execute() saw: ('book a flight', 'request-1', 'WAT-request-1')
execute() saw: ('Friday', 'request-2', 'WAT-request-2')

The same result comes from a real uvicorn server with serve_a2a, and from message/stream.

Proposed fix

Snapshot the request's contextvars where BedrockAgentCoreContext is already set per request, and have build_a2a_app run the user's executor in the snapshot of the message being handled:

# BedrockCallContextBuilder.build(), after the BedrockAgentCoreContext.set_* calls
call_context.state[_CONTEXTVARS_KEY] = contextvars.copy_context()


class _PerMessageContextExecutor(AgentExecutor):
    """Runs the wrapped executor in the contextvars of the request that carried the message."""

    def __init__(self, inner: AgentExecutor):
        self._inner = inner
        self.agent = getattr(inner, "agent", None)  # keeps the auto-generated agent card's name and description

    async def execute(self, context, event_queue):
        snapshot = context.call_context.state.get(_CONTEXTVARS_KEY) if context.call_context else None
        if snapshot is None:
            return await self._inner.execute(context, event_queue)
        # cancelling the task worker cancels the awaited child too
        return await snapshot.run(asyncio.create_task, self._inner.execute(context, event_queue))

    async def cancel(self, context, event_queue):
        return await self._inner.cancel(context, event_queue)  # already runs in the cancel request's context


# build_a2a_app(): wrap the executor (harmless on a2a-sdk 0.3.x, where execute() already runs per request)
executor = _PerMessageContextExecutor(executor)

Notes:

• This is the same pattern the MCP Python SDK adopted for the same reason. In modelcontextprotocol/python-sdk#2298 (merged; fixes #1969), streams capture the sender's context at send(), and the client transport's post_writer restores it per message.

• snapshot.run(asyncio.create_task, ...) works on Python 3.10+. create_task(..., context=) needs 3.11+.

• Verified behaviour with this wrapper:

  • per-message values for both send and stream, under both v1 and v0.3 method names

  • no cross-task bleed across 40 concurrent tasks

  • identical error and task-store behaviour

  • cancellation propagates to the child

  • direct-Message replies and single-turn tasks unchanged

  • harmless on a2a-sdk 0.3.7

• One limit: only execute() is covered. a2a-sdk's background bookkeeping (task-store writes, push notifications) still runs in the worker's context.

• A custom context_builder passed to build_a2a_app would need to call super().build() (or set the key itself) to keep the behaviour; that is worth a line in the docstring.

• An upstream a2a-sdk fix (capturing the sender context in ActiveTask.enqueue_request) would also work. a2a-python treats ServerCallContext.state as the per-request channel by design ([#94](feat: Introduce a ServerCallContext a2aproject/a2a-python#94)) and declined a similar request for trace context ([#281]([Feat]: EventQueue should propagate trace context? a2aproject/a2a-python#281)). Since build_a2a_app is where BedrockAgentCoreContext is set, fixing it here also fixes AgentCore's own context.

Workaround for users today

The same fix works from user code, using only build_a2a_app(context_builder=...) and an executor wrapper:

import asyncio, contextvars
from a2a.server.agent_execution import AgentExecutor
from bedrock_agentcore.runtime import BedrockCallContextBuilder, build_a2a_app


class PerMessageContextBuilder(BedrockCallContextBuilder):
    def build(self, request):
        ctx = super().build(request)
        ctx.state["contextvars"] = contextvars.copy_context()
        return ctx


class PerMessageContext(AgentExecutor):
    def __init__(self, inner):
        self._inner = inner
        self.agent = getattr(inner, "agent", None)

    async def execute(self, context, event_queue):
        snapshot = context.call_context.state["contextvars"]
        await snapshot.run(asyncio.create_task, self._inner.execute(context, event_queue))

    async def cancel(self, context, event_queue):
        return await self._inner.cancel(context, event_queue)


app = build_a2a_app(PerMessageContext(MyExecutor()), context_builder=PerMessageContextBuilder())

Both pieces are required. serve_a2a takes the same context_builder=. build_a2a_app(context_builder=) exists from bedrock-agentcore 1.4.7, and a2a-sdk 1.x support from 1.19.0.

With it, the reproduction prints ('Friday', 'request-2', 'WAT-request-2') for the second turn.

Related

• feat(a2a): migrate runtime integration to a2a-sdk v1 #591, the migration to a2a-sdk v1, where this behaviour started (a2a-sdk 0.3.x ran the executor per request)

• OTEL context token detached across asyncio/thread boundary in memory client + Strands session_manager #456, OTel context across asyncio/thread boundaries

• a2a-python [#1296]([Bug]: DefaultRequestHandlerV2 keeps the ActiveTask (producer, consumer, 2 dispatchers) alive forever after a direct Message or input-required response a2aproject/a2a-python#1296), ActiveTask kept alive after input-required, which lengthens the window