A2A with agent engine session service in api_server runtime

[guillaumeblaquiere]

Describe the bug
It's not possible to use agent-engine session service with A2A

To Reproduce
Run your ADK agent like this
adk api_server --a2a --host 0.0.0.0 --port 8080 --session_service_uri "agentengine://${AGENT_ENGINE_ID}" --memory_service_uri "agentengine://${AGENT_ENGINE_ID}"

You get this error when you use the A2A Endpoint (I use it with Agent Space (a.k.a Gemini Enterprise))

Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/google/adk/a2a/executor/a2a_agent_executor.py", line 167, in execute
    await self._handle_request(context, event_queue)
  File "/usr/local/lib/python3.12/site-packages/google/adk/a2a/executor/a2a_agent_executor.py", line 208, in _handle_request
    session = await self._prepare_session(context, run_request, runner)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/google/adk/a2a/executor/a2a_agent_executor.py", line 303, in _prepare_session
    session = await runner.session_service.get_session(
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/google/adk/sessions/vertex_ai_session_service.py", line 162, in get_session
    get_session_response, events_iterator = await asyncio.gather(
                                            ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/vertexai/_genai/sessions.py", line 882, in get
    response = await self._api_client.async_request(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/google/genai/_api_client.py", line 1365, in async_request
    result = await self._async_request(
             ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/google/genai/_api_client.py", line 1310, in _async_request
    return await self._async_retry(  # type: ignore[no-any-return]
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/tenacity/asyncio/__init__.py", line 111, in __call__
    do = await self.iter(retry_state=retry_state)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/tenacity/asyncio/__init__.py", line 153, in iter
    result = await action(retry_state)
             ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/tenacity/_utils.py", line 99, in inner
    return call(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/tenacity/__init__.py", line 420, in exc_check
    raise retry_exc.reraise()
          ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/tenacity/__init__.py", line 187, in reraise
    raise self.last_attempt.result()
          ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/concurrent/futures/_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.12/site-packages/tenacity/asyncio/__init__.py", line 114, in __call__
    result = await fn(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/google/genai/_api_client.py", line 1290, in _async_request_once
    await errors.APIError.raise_for_async_response(client_response)
  File "/usr/local/lib/python3.12/site-packages/google/genai/errors.py", line 166, in raise_for_async_response
    raise ClientError(status_code, response_json, response)
google.genai.errors.ClientError: 400 INVALID_ARGUMENT. {'error': {'code': 400, 'message': 'List of found errors:\t1.Field: name; Message: Invalid Session resource name.\t', 'status': 'INVALID_ARGUMENT', 'details': [{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'name', 'description': 'Invalid Session resource name.'}]}]}}

If you start the agent without the session and memory service
adk api_server --a2a --host 0.0.0.0 --port 8080

You will use the InMemoryService and it works well

Expected behavior
I expect no error when I use AgentEngine memory service and A2A.

A2A invocation belonging to an external process/flow/session management, I would assume that the A2A request do not need persistent session service. So, having InMemory session service only for the A2A endpoint is fine in my opinion

Desktop (please complete the following information):

• Python version(python -V): 3.12.11
• ADK version(pip show google-adk): 1.18.0

Model Information:

• Are you using LiteLLM: No
• Which model is being used(e.g. gemini-2.5-pro): gemini-2.5-flash

Additional context
I would imagine solving the issue here

And I imagine a runner update like this.

      async def _get_a2a_runner_async() -> Runner:
          original_runner = await adk_web_server.get_runner_async(captured_app_name)
          runner = copy.copy(original_runner)  # Create a shallow copy
          runner.memory_service = InMemoryMemoryService()
          runner.session_service = InMemorySessionService()
          runner.artifact_service = InMemoryArtifactService()
          runner.credential_service = InMemoryCredentialService()
          return runner

        return _get_a2a_runner_async

What do you think?

[narmada-nannaka]

Alternative Solution: FirestoreSessionService for A2A Persistence

Hi @guillaumeblaquiere,

I've been exploring the session service + A2A integration challenge and developed a potential solution using Firestore as a custom session backend.

Context

I'm working on a multi-agent system where:

• Orchestrator agent delegates to multiple sub-agents + 1 remote A2A agent
• Both services need persistent sessions for state management
• Sessions must be retrievable across service restarts

Approach

I implemented a FirestoreSessionService(BaseSessionService) that:

• ✅ Implements all required async session methods
• ✅ Works with RemoteA2aAgent
• ✅ Supports custom session IDs
• ✅ Handles document-based storage with Firestore
• ✅ Provides session isolation per user

Simplified Example

class FirestoreSessionService(BaseSessionService):
    """Firestore-backed session service compatible with A2A workflows."""
    
    def __init__(self, firestore_client, collection_name="adk-sessions"):
        self.db = firestore_client
        self.collection_name = collection_name
    
    async def create_session(self, session_id, app_name, user_id, **kwargs):
        session = Session(id=session_id, app_name=app_name, 
                         user_id=user_id, state=kwargs.get('state', {}))
        
        doc_ref = self.db.collection(self.collection_name).document(session_id)
        doc_ref.set({
            "session_id": session_id,
            "app_name": app_name,
            "user_id": user_id,
            "state": session.state,
            "created_at": datetime.now(timezone.utc),
            "updated_at": datetime.now(timezone.utc)
        })
        return session
    
    async def get_session(self, *, app_name, user_id, session_id, config=None):
        doc_ref = self.db.collection(self.collection_name).document(session_id)
        doc = doc_ref.get()
        
        if not doc.exists:
            return None
            
        data = doc.to_dict()
        return Session(
            id=data["session_id"],
            app_name=data["app_name"],
            user_id=data["user_id"],
            state=data.get("state", {})
        )
    
    # ... update_session, delete_session, list_sessions methods

Usage with A2A

# Initialize Firestore session service
db = firestore.Client(project="my-project")
session_service = FirestoreSessionService(db, collection_name="agent-sessions")

# Works with RemoteA2aAgent
remote_agent = RemoteA2aAgent(agent_card=url, httpx_client=client)
orchestrator = MyOrchestrator(sub_agents=[local_agent, remote_agent])

runner = Runner(
    agent=orchestrator,
    session_service=session_service  # Persistent across A2A calls
)

Proposal

Rather than limiting A2A to InMemorySessionService, would maintainers be open to a PR contributing FirestoreSessionService as an alternative? This would:

1. Provide persistent sessions for A2A workflows
2. Enable state persistence across service restarts
3. Support multi-instance deployments
4. Address Agent-as-a-Tool uses InMemorySessionService #2754 and aspects of Session State Leakage: Multiple Sessions Showing Accumulated Data #3513 as well

I can prepare:

• Full implementation following ADK patterns
• Comprehensive test suite (mocked Firestore)
• Example in contributing/samples/firestore_session_service/
• Documentation on use cases

Would this be a useful addition? Happy to adapt to maintainer preferences or contribute as an example/pattern instead of core library if preferred.

[guillaumeblaquiere]

Hello @narmada-nannaka

Thank you for this proposal that forced me to deep dive in the A2A protocol.

As outcome, I have 2 remarks:

• Firstly, the storage layer is not the problem (Firestore, DB, Agent Engine), the problem is the session management. When you receive a Task and a Context in A2A, it's out of any session previously created. Therefore, when you are doing this

   async def get_session(self, *, app_name, user_id, session_id, config=None):
        doc_ref = self.db.collection(self.collection_name).document(session_id)
        doc = doc_ref.get()
        
        if not doc.exists:
            return None
            
       .....

If the session is not found in Firestore, you return None, that will lead to the same error as mine.

• Secondly, the A2A Protocol mention this

...
Agents MAY use the contextId to maintain internal state, conversational history, or LLM context across multiple interactions
...

According to a contextID provided by the client, the agent MAY maintain the context and state. not a SHOULD or a MUST

So, my solution is not perfect, but finally acceptable and compliant with the protocol.

---

I think the best way to solve this would be to create automatically a session if the A2A protocol is active and the session not found. This solution won't work with AgentEngine session service because you can't define your own sessionID (here will be the ContextID provided by the client, if any, could be missing). But should work with non-managed session persistence services.

Eventually, it's only my opinion, and I'm not a maintainer, only a contributor. Maybe the team will have other remarks and point of view on your proposal!

[narmada-nannaka]

Thank you for the detailed analysis, @guillaumeblaquiere!

You're absolutely right - I missed the critical auto-creation requirement for A2A workflows. I see the distinction now:

Two A2A Patterns

1. External A2A (your case): Client sends arbitrary contextId

• You've never seen this contextId before
• Session doesn't exist in your service
• Need to handle gracefully

2. Internal A2A (my case): I control both ends

• I create the session before the A2A call
• Both services share the same session store
• Session already exists when A2A agent looks it up

Solutions for Each Pattern

Your InMemory solution is perfect for Pattern 1 (stateless, external clients).

My Firestore proposal could add value for:

• Pattern 1 when you need audit trails (auto-create on first call)
• Pattern 2 when you need cross-service state (pre-created sessions)
• Multi-turn A2A conversations (persist context between calls)

Implementation Consideration

For auto-creating sessions with external clients, I'd suggest adding:

• TTL (Time-To-Live): Auto-cleanup of ephemeral sessions to prevent database bloat
• Rate Limiting: Protect against session spam/DoS scenarios

Example:

class FirestoreSessionService(BaseSessionService):
    def __init__(self, firestore_client, auto_create_ttl_hours=24):
        self.db = firestore_client
        self.auto_create_ttl = auto_create_ttl_hours
    
    async def get_session(self, *, session_id, ...):
        doc = self.db.collection("sessions").document(session_id).get()
        
        if not doc.exists:
            # Auto-create with expiration
            expiry = datetime.now(timezone.utc) + timedelta(hours=self.auto_create_ttl)
            return await self.create_session(
                session_id=session_id,
                state={"expires_at": expiry.isoformat(), "source": "a2a_auto"}
            )
        
        return Session(...)

Would this be useful as an additional option alongside InMemory, or do you think the simplicity of InMemory makes it the better default for external A2A?

Looking forward to maintainer input on this!

[guillaumeblaquiere]

I agree with you, my patch is too agressive. I should only set in-memory if it's with AgentEngine. Other solution would have the ability to create session with the contextID.

The problem with AgentEngine session service is that you can't provide sessionID (here the contextID) and you can't make the link between the subsequent calls. it's a lack in the current Google Cloud API. But DB session service, In memory, Firestore (...) does not have this limitation.

[surajksharma07]

The issue occurs because Agent Engine's session service doesn't support user-provided session IDs, but A2A endpoints map the context_id directly to session_id (see vertex_ai_session_service.py:103-107 and a2a/converters/request_converter.py:78-117).

Your proposed solution is correct - conditionally override the session service with InMemorySessionService for A2A endpoints when Agent Engine is configured.

This is compliant with the A2A protocol since agents MAY (not MUST) maintain state based on context_id.

Workaround: Remove the --session_service_uri and --memory_service_uri flags when running adk api_server --a2a, which defaults to InMemorySessionService and works correctly.

[surajksharma07]

@seanzhou1023 Please have a look into it.