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?