A2A with agent engine session service in api_server runtime

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?

@guillaumeblaquiere

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:

Provide persistent sessions for A2A workflows
Enable state persistence across service restarts
Support multi-instance deployments
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.

@narmada-nannaka

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!

@guillaumeblaquiere

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!

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.

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.

@seanzhou1023

@seanzhou1023 Please have a look into it.

The bug still exists in v1.23.0 - fast_api.py:491-494 still uses the configured session service for A2A endpoints, which fails with Agent Engine.

PR #3587 was not merged.

As a workaround, remove --session_service_uri and --memory_service_uri flags when running adk api_server --a2a to use InMemorySessionService by default.

@surajksharma07

@surajksharma07 It's not a valid workaround if your agent must work with session when used by API, and work also with A2A. I do not want to deploy 2 versions of the same agent only for a session persistence issue.

thanks for flagging this, since v1.29.0 the vertex ai session service forwards a caller-supplied session id, which should let a2a create the session under the incoming context id. closing as stale, open with a fresh traceback if it still fails.

fix: agent engine memory service with A2A endpoint activated

Navigation Menu

Repository navigation

A2A with agent engine session service in api_server runtime #3458

Alternative Solution: FirestoreSessionService for A2A Persistence

Context

Approach

Simplified Example

Usage with A2A

Proposal

Thank you for the detailed analysis, @guillaumeblaquiere!

Two A2A Patterns

Solutions for Each Pattern

Implementation Consideration

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

A2A with agent engine session service in api_server runtime #3458

Description

Activity

narmada-nannaka commented on Nov 28, 2025

Alternative Solution: FirestoreSessionService for A2A Persistence

Context

Approach

Simplified Example

Usage with A2A

Proposal

guillaumeblaquiere commented on Nov 28, 2025

narmada-nannaka commented on Dec 1, 2025

Thank you for the detailed analysis, @guillaumeblaquiere!

Two A2A Patterns

Solutions for Each Pattern

Implementation Consideration

guillaumeblaquiere commented on Dec 10, 2025

surajksharma07 commented on Dec 14, 2025

surajksharma07 commented on Dec 14, 2025

surajksharma07 commented on Feb 5, 2026

guillaumeblaquiere commented on Apr 2, 2026

GWeale commented on Aug 6, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions