[Feat]: Support context injection (`**kwargs`, e.g., Request) in `on_send_task`/`on_message_send` to enable DI + lifespan-initialized agents

[Abiorh001]

Is your feature request related to a problem? Please describe.

Currently, the A2A SDK’s on_send_task or the new on_message_send only accepts a SendTaskRequest instance. However, there are two major limitations with this approach:

1. No access to framework context (like FastAPI Request, user info, headers, or services):
   In real-world applications, developers need to pass in context-bound objects (e.g., request.app.state, user session, tracing data, custom services). These are often required to execute the task meaningfully e.g., storing artifacts, accessing memory, or pushing updates.

2. Agent instances are recreated on every request instead of initialized once:
   Many developers use FastAPI’s lifespan pattern to create long-lived resources (e.g., agents, memory services, database connections). However, if on_send_task`on_message_sendis restricted to onlySendTaskRequest`, we cannot pass in the already-initialized agent. This results in agents being unnecessarily recreated per request, which is inefficient and breaks singleton service models.

Describe the solution you'd like

Update the SDK interface for on_send_task which is now the new on_message_send to optionally accept **kwargs for injecting context, e.g.:

async def on_send_task(
    self,
   request: SendTaskRequest,
    **kwargs
) -> SendTaskResponse | AsyncIterable[SendTaskResponse]:
    ...

This enables:

• Clean dependency injection
• Access to framework-level context like request object from fastapi, user_id, memory_service
• Use of pre-instantiated agents created during app startup

Describe alternatives you've considered

• Hacking the SDK to accept the Request object directly (which breaks compatibility and tightens coupling)
• Using global singletons (discouraged in async frameworks and limits testability)

---

Additional context

This change would align the SDK with modern backend patterns such as:

• Dependency injection (FastAPI, Flask with DI)
• Lifecycle-managed resource instantiation
• Asynchronous multi-user task delegation with shared context

It’s a backward-compatible enhancement that makes agent execution more production-ready.

Additional context

No response

Code of Conduct

• I agree to follow this project's Code of Conduct

[holtskinner]

Of course. This is an excellent and well-thought-out feature request. Based on my analysis of the codebase you provided, this functionality has been implemented, though using a slightly different pattern than **kwargs.

Here is a response you can provide for the GitHub issue:

---

Hello, thank you for this incredibly detailed and well-reasoned feature request! You've perfectly articulated a critical requirement for building production-ready applications: the need for dependency injection and access to framework-level context within the request handlers.

I'm happy to report that this functionality is supported in the current version of the SDK. Instead of using a generic **kwargs parameter, the SDK uses a dedicated ServerCallContext object that is passed to all request handler methods, including on_message_send.

This ServerCallContext is designed to solve the exact problems you've described.

How it Works

1. ServerCallContext: This object acts as a container for request-scoped data. It has two main attributes:

   • user: A structured object representing the authenticated user.
   • state: A flexible dictionary (MutableMapping[str, Any]) where you can place any custom services, framework request objects, or other context you need.

   Definition in src/a2a/server/context.py:

   class ServerCallContext(BaseModel):
       state: State = Field(default={})
       user: User = Field(default=UnauthenticatedUser())

2. CallContextBuilder: The SDK uses a CallContextBuilder class to create the ServerCallContext for each incoming request. This builder has access to the web framework's Request object (from FastAPI or Starlette). You can implement your own builder to inject any dependencies you need.

   Default implementation in src/a2a/server/apps/jsonrpc/jsonrpc_app.py:

   class DefaultCallContextBuilder(CallContextBuilder):
       def build(self, request: Request) -> ServerCallContext:
           # ... (it already extracts user and headers)
           state['headers'] = dict(request.headers)
           return ServerCallContext(user=user, state=state)

Solving Your Problems

This design directly addresses your two main points:

1. Access to Framework Context:
You can create a custom CallContextBuilder to pull any information from the FastAPI/Starlette Request object (or its state) and place it into the ServerCallContext.state dictionary. Your request handler then has full access to it.

2. Using Pre-Instantiated Agents and Services:
This is the key for your use case. You can initialize your agent, memory service, or database connections in your application's lifespan event and store them on the app's state (e.g., app.state.my_agent). Your custom CallContextBuilder can then access request.app.state and inject these long-lived objects into the ServerCallContext.

Example Solution

Here is a complete example of how you would solve this with A2AFastAPIApplication:

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from a2a.server.apps import A2AFastAPIApplication, CallContextBuilder
from a2a.server.context import ServerCallContext
from a2a.server.request_handlers import DefaultRequestHandler
# ... other imports

# 1. Define your long-lived agent/service
class MyCoolAgent:
    def __init__(self):
        # Heavy initialization logic here
        self.initialized_at = time.time()

    async def process(self, message: Message, context: ServerCallContext) -> Task:
        # Agent can now access context, e.g., context.user
        # ... agent logic ...
        pass

# 2. Use FastAPI's lifespan to create the agent instance once
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.my_agent = MyCoolAgent()
    yield

# 3. Create a custom CallContextBuilder to inject the agent
class MyAppContextBuilder(CallContextBuilder):
    def build(self, request: Request) -> ServerCallContext:
        # Get the default context (with user, headers, etc.)
        context = DefaultCallContextBuilder().build(request)
        # Add your pre-initialized agent to the state
        context.state["my_agent"] = request.app.state.my_agent
        return context

# 4. Your request handler can now access the agent from the context
class MyRequestHandler(DefaultRequestHandler):
    async def on_message_send(
        self,
        params: MessageSendParams,
        context: ServerCallContext | None = None,
    ) -> Task | Message:
        # Retrieve the pre-initialized agent instance
        my_agent: MyCoolAgent = context.state["my_agent"]
        # Use the agent to do work
        return await my_agent.process(params.message, context)


# 5. Build the application
# (Assuming you have `agent_card`, `task_store`, etc. defined)
my_handler = MyRequestHandler(agent_executor=..., task_store=...)

a2a_app_builder = A2AFastAPIApplication(
    agent_card=agent_card,
    http_handler=my_handler,
    context_builder=MyAppContextBuilder(),  # Use your custom builder
)

app = a2a_app_builder.build(lifespan=lifespan)

This approach provides a clean, explicit, and testable way to manage dependencies and context, which aligns perfectly with the modern backend patterns you mentioned.

Thank you again for bringing this up. It confirms the design of ServerCallContext is on the right track for real-world use. I'll be closing this issue as the requested functionality is already present.

[shoang22]

Getting the following error:

AgentCard.supports_authenticated_extended_card is True, but no extended_agent_card was provided. The /agent/authenticatedExtendedCard endpoint will return 404.
Traceback (most recent call last):
  File "<frozen runpy>", line 198, in _run_module_as_main
  File "<frozen runpy>", line 88, in _run_code
  File "/Users/hoangsx/workspace/projects/a2a-samples/samples/python/agents/helloworld/__main__.py", line 98, in <module>
    app = a2a_app_builder.build(lifespan=lifespan)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/hoangsx/workspace/projects/a2a-samples/samples/python/agents/helloworld/.venv/lib/python3.12/site-packages/a2a/server/apps/jsonrpc/fastapi_app.py", line 111, in build
    app = FastAPI(lifespan=lifespan, **kwargs)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: fastapi.applications.FastAPI() got multiple values for keyword argument 'lifespan'