Deploying an A2A Agent with Auth & Session Handling

[djanito]

Hi 👋

I’m trying to deploy an A2A agent using Google ADK, and I’m struggling to find clear documentation around authentication, token propagation, and session handling.

🎯 What I Need

I would like to:

1. Handle authentication to access the agent (JWT-based).
2. Propagate the auth token to custom remote MCP servers.
3. Access decoded JWT payload fields inside my tools.
4. Let the agent handle sessions automatically (A2A-native sessions).

---

✅ What I Implemented So Far

1️⃣ Convert Agent to A2A
I converted my root agent to A2A using to_a2a:

def create_app(root_agent) -> Optional[object]:
    try:
        from google.adk.a2a.utils.agent_to_a2a import to_a2a

        logger.info(f"Creating A2A application for {config.SERVICE_NAME}")

        # Convert agent to A2A protocol
        app = to_a2a(
            root_agent,
            port=config.SERVICE_PORT,
            host=config.SERVICE_HOST,
        )

        app.routes.append(Route("/health", health_check, methods=["GET"]))
        app.add_middleware(JWTAuthMiddleware)

        logger.info(
            f"A2A application created successfully on {config.SERVICE_HOST}:{config.SERVICE_PORT}"
        )
        return app

    except ImportError as e:
        logger.warning(f"Server mode dependencies not available: {e}")
        logger.warning("Agent will only work in local/CLI mode")
        return None

2️⃣ Custom JWT Middleware

I implemented a JWTAuthMiddleware using ContextVar to propagate the decoded payload across the agent execution:

from contextvars import ContextVar
request_context = ContextVar("request_context", default=None)

class JWTAuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(
        self,
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ):
        """Process incoming request with JWT authentication."""
        try:
            token = self.extract_token_from_request(request)
            payload = self.verify_token(token)
        except Exception as exc:
            return self._handle_authentication_exception(request, exc)

        request_context.set({"token": token, "payload": payload})
        request.state.jwt_token = token
        return await call_next(request)

3️⃣ Inject User Info into Agent Context

I added a before_agent_callback to push user info into the agent state:

async def _before_agent_callback(callback_context: CallbackContext):
    info = request_context.get()
    if info:
        payload = info.get("payload", {})
        user_profile = payload.get("profile", {})
        callback_context.state["user:email"] = user_profile.get("email")

---

With this setup. I'm having some issues:

• When running the agent locally using adk web to test it, the jwt middleware is bypassed.
• When running the server using uvicorn my_agent.agent:app --reload --host 0.0.0.0 --port 8080, I need to send call using A2A protocol and that doesn't handle the sessions.

🤔 Main Questions

1. Is using a ContextVar + before_agent_callback the correct pattern for propagating JWT payload into tools?
2. Is there a more “ADK-native” way to handle authentication?
3. What is the recommended way to:
   • Validate JWT
   • Inject decoded payload into agent state
   • Propagate token to remote MCP tools
4. How should auth integrate with A2A session handling?
5. Is adk web compatible with custom FastAPI middleware, or is it intended only for local/dev use?

[musaabhasan]

The pattern I would use is to keep these as three separate concerns: edge authentication, ADK state, and downstream tool/MCP authorization.

1. Validate the JWT at the ASGI/FastAPI boundary in production. adk web is mainly a local/dev runner, so I would not expect your custom FastAPI middleware to be exercised there in the same way as uvicorn my_agent.agent:app. For local testing, inject a mock user into state or run the actual ASGI app.

2. A ContextVar can work if the request and agent execution stay in the same async context, but I would treat it as a bridge, not the source of truth. In the callback, copy only the minimum identity/authorization claims into callback_context.state, for example user id, tenant id, email, groups/scopes, and a correlation id. I would avoid placing the raw JWT in agent-visible state because it may become part of logs, traces, or model context depending on later code.

3. For remote MCP servers, pass credentials at the tool/client boundary, not through the prompt. The tool wrapper or MCP client should read the validated request identity and attach the bearer token or exchange it for a downstream token with the scopes that MCP server needs. If the MCP server acts on behalf of the user, use delegated scopes; if it is a service integration, use a service principal and enforce user-level authorization before the tool call.

4. For sessions, bind the A2A session/user identifiers to the authenticated principal. Authentication should be checked on every request, while session state should store non-secret continuity data. Do not rely on a session id alone as authorization.

So your direction is reasonable, but I would tighten it by keeping raw tokens out of agent state, using middleware only in the production ASGI path, and testing adk web with explicit mock identity instead of expecting the same middleware path.

If this resolves the deployment pattern, please mark it as the answer so others can find it.