feat(api_server): record a CallerPrincipal on /run and /run_sse

docs/guides/README.md

@@ -12,6 +12,7 @@ This directory contains specific developer guides for the ADK Python implementat
 
 ### Agents
 * [BaseAgent](agents/base_agent/index.md) - The foundational base class for custom agents, container orchestrators, and lifecycle callbacks.
+* [CallerPrincipal](agents/caller_principal/index.md) - The caller identity a serving layer establishes for an invocation, and how tool confirmation uses it.
 * [Context](agents/context/index.md) - The runtime interface for state, artifacts, memory, credentials, and dynamic execution.
 * [Creating Agents with Configurations](agents/config/index.md) - Building and wiring multi-agent graphs from external YAML configuration files.
 * [InvocationContext](agents/invocation_context/index.md) - The runtime dependency container and execution state for a single invocation turn.

docs/guides/agents/caller_principal/index.md

@@ -0,0 +1,237 @@
+# CallerPrincipal
+
+`CallerPrincipal` records whether a serving layer authenticated the caller of
+an invocation, and as whom. The human-in-the-loop tool confirmation flow reads
+it to decide whether an approval that arrived over the wire is one the agent
+may act on.
+
+## Introduction
+
+A tool that requires confirmation pauses the agent until a user approves the
+call. When the agent is served remotely, the approval arrives as a message on
+the same channel as everything else the remote caller sends, so without a
+principal the framework cannot tell an operator's approval from a remote peer
+approving the dangerous tool call it just caused. `CallerPrincipal` carries the
+one fact that decides this: whether the layer that received the request
+verified who sent it.
+
+The principal lives on `InvocationContext.caller_principal`. `Runner.run_async`
+accepts it as a keyword argument, and the A2A executor sets it for every
+request it serves, so an application that serves agents over A2A gets it
+without writing any code. An application with its own serving layer sets it
+from its own authenticator.
+
+The design keeps the principal separate from the transport on purpose. A
+transport is a proxy for identity, and a proxy for identity fails in both
+directions: it refuses authenticated callers because of how they connected,
+and it admits anyone who reaches a path the proxy does not cover. Asking the
+authentication question directly avoids both failures.
+
+## Get started
+
+This example serves an agent behind an application's own request handler. The
+handler runs its authenticator first, then tells the runner what it found. A
+confirmation-gated tool is included so that the principal has something to
+decide.
+
+```python
+from google.adk.agents import CallerPrincipal
+from google.adk.agents import LlmAgent
+from google.adk.runners import InMemoryRunner
+from google.adk.tools.bash_tool import ExecuteBashTool
+
+agent = LlmAgent(name="ops_agent", tools=[ExecuteBashTool()])
+runner = InMemoryRunner(agent=agent, app_name="ops")
+
+
+async def handle_request(request, new_message):
+  # Your authenticator runs before this point. Report what it established;
+  # do not infer anything from the message the caller sent.
+  user = request.authenticated_user
+  principal = CallerPrincipal(
+      authenticated=user is not None,
+      user_name=user.name if user else None,
+      source="gateway",
+  )
+  async for event in runner.run_async(
+      user_id=request.user_id,
+      session_id=request.session_id,
+      new_message=new_message,
+      caller_principal=principal,
+  ):
+    yield event
+```
+
+An in-process caller, such as a test or a command-line tool that calls
+`runner.run_async` directly, does not pass a principal. Leaving it unset says
+that no remote trust boundary was crossed, which is true for that caller.
+
+## How it works
+
+`InvocationContext.caller_principal` has three meaningful states, and the tool
+confirmation flow treats each differently.
+
+| State | Meaning | Tool confirmation |
+| :--- | :--- | :--- |
+| `None` | The invocation started in process. Nothing had to vouch for the caller, because the caller is the operator. | Honored. |
+| `authenticated=False` | A serving layer handled the request and could not say who sent it. | Warned about by default; refused when strict mode is on. |
+| `authenticated=True` | The serving layer verified the caller. | Honored. |
+
+A refusal is explicit rather than silent. The confirmation is delivered to the
+tool with `confirmed=False`, which is the same state a user's decline produces,
+so a tool that follows the confirmation contract returns its rejection response
+and the turn ends with a reason the caller can see. The pending call never
+hangs waiting for an approval that will not come.
+
+The same decision applies wherever a confirmation answer enters the framework:
+a confirmation carried in a user event to an `LlmAgent`, and an answer that
+resumes a paused workflow node (`RequestInput` with interrupt id
+`wf_tool_confirmation:<node_path>`), whether the node wraps a tool or a
+function declared with `@node(require_confirmation=True)`. In a workflow node
+the answer is rewritten before it is stored on the context, so tools that read
+`tool_confirmation` themselves, such as `ExecuteBashTool`, see the refusal too.
+
+The principal is derived only from what the serving layer established. It is
+never read from message content, from event authorship, or from transport
+metadata, because a remote caller controls all three and could set any of them
+to whatever the framework wanted to see.
+
+### A2A
+
+`A2aAgentExecutor` builds the principal from the A2A server's call context on
+every request. The caller is authenticated exactly when the A2A server
+authenticated it, which is also the case in which the invocation's user id is
+the authenticated user name rather than a generated `A2A_USER_` value. The two
+cannot disagree about the same request.
+
+An A2A server that runs without an authenticator produces an unauthenticated
+principal for every request. With the default settings that yields a warning
+per approval and the approval is honored, which is the behavior those
+deployments had before the principal existed.
+
+### ADK API server
+
+The `/run` and `/run_sse` routes of the ADK API server (`adk api_server`,
+`adk web`, `get_fast_api_app`) record a principal on every invocation as
+well. The server does not authenticate callers itself, and the `user_id` in
+the request body is a claim the caller made rather than one the server
+checked, so without anything in front of these routes every request runs as
+`CallerPrincipal(authenticated=False, source="api_server")`. With the default
+settings that is a warning per approval, as for an unauthenticated A2A caller.
+
+A deployment that authenticates callers in front of these routes, with an
+identity-aware proxy, a gateway or its own middleware, states the result on
+`request.state.caller_principal`, and the server uses that value as is:
+
+```python
+from google.adk.agents import CallerPrincipal
+from google.adk.cli.fast_api import get_fast_api_app
+
+app = get_fast_api_app(agents_dir="./agents", web=False)
+
+
+@app.middleware("http")
+async def record_caller(request, call_next):
+  # Your authenticator decides who sent this request; report that outcome.
+  user = verify_request(request)  # None when the request is not authenticated.
+  if user is not None:
+    request.state.caller_principal = CallerPrincipal(
+        authenticated=True, user_name=user.name, source="gateway"
+    )
+  return await call_next(request)
+```
+
+Under strict mode, approvals arriving on `/run` or `/run_sse` are refused
+unless the deployment sets the principal this way. The dev UI served by `adk
+web` has no authentication of its own, so run it without strict mode, or add
+middleware that recognizes its operator.
+
+### Strict mode
+
+Strict mode refuses confirmations from unauthenticated callers instead of
+warning about them. It is the `STRICT_CALLER_PRINCIPAL` feature and is off by
+default, because turning it on changes what an existing deployment does with an
+approval, and an upgrade should not do that on its own. The intent is to make
+strict mode the default at the next major version.
+
+Turn it on with the environment variable ADK uses for every feature:
+
+```bash
+export ADK_ENABLE_STRICT_CALLER_PRINCIPAL=1
+```
+
+Or programmatically, when environment variables are not practical in your
+deployment:
+
+```python
+from google.adk.features import FeatureName
+from google.adk.features import override_feature_enabled
+
+override_feature_enabled(FeatureName.STRICT_CALLER_PRINCIPAL, True)
+```
+
+Strict mode never affects an invocation with no principal. An in-process
+caller crossed no boundary, so there is nothing to refuse.
+
+## Configuration options
+
+`CallerPrincipal` is a frozen model with three fields. It rejects unknown
+fields, so a typo cannot silently produce a principal that says nothing.
+
+| Option | Type | Default | Description |
+| :--- | :--- | :--- | :--- |
+| `authenticated` | `bool` | required | Whether a serving-layer authenticator verified the caller. |
+| `user_name` | `str \| None` | `None` | The verified identity when `authenticated` is `True`. |
+| `source` | `str \| None` | `None` | A short label for the serving edge that set the principal. |
+
+`authenticated` is the only field a trust decision reads. Set it to `True` only
+when your authenticator actually verified the caller. A request that carries a
+name it asserted about itself, without verification, is not authenticated, and
+saying otherwise reopens the problem the principal exists to close.
+
+`user_name` is informational. It is the identity the authenticator established,
+kept alongside the decision so that a log line or an audit record can say who
+approved a tool call. Leave it `None` when `authenticated` is `False`; a name
+without verification behind it is a claim, not an identity.
+
+`source` is a label such as `"a2a"` or `"gateway"` that names the edge that
+set the principal. It appears in the confirmation flow's log messages so that
+an operator reading a warning can tell which entry point produced it. Nothing
+makes a trust decision on it, so a transport cannot become a proxy for identity
+through this field either.
+
+## Advanced applications
+
+### Custom A2A request converters
+
+`A2aAgentExecutorConfig.request_converter` lets an application replace the
+function that turns an A2A request into a run request. The executor sets the
+principal from the server's call context after the converter returns,
+regardless of what the converter produced. A converter can therefore neither
+drop the principal nor assert one the server never established, which keeps
+the trust decision structural rather than dependent on every converter
+remembering to make it.
+
+### Serving layers with more than one entry point
+
+An application that exposes an agent through several edges can give each a
+distinct `source` and let them share the same rule for `authenticated`. The
+confirmation flow's log messages then identify the edge, and the trust
+decision stays uniform across all of them.
+
+## Limitations
+
+- The principal describes the caller of the invocation as a whole. ADK does
+  not verify the identity behind each individual event within a session.
+- Only tool confirmation answers are gated on the principal today, on every
+  path that accepts one: the `LlmAgent` confirmation flow and the resume of a
+  workflow tool node or `require_confirmation` function node. Other consumers
+  can read `InvocationContext.caller_principal`, but the framework does not yet
+  gate anything else on it.
+- The synchronous `Runner.run` wrapper and live sessions do not accept a
+  principal. Both behave as in-process callers.
+- Setting a principal does not authenticate anything by itself. The ADK API
+  server's `/run` and `/run_sse` routes record an unauthenticated principal
+  unless the deployment's own authentication sets one; the principal records
+  the outcome of authentication, it does not perform it. `/run_live` sessions
+  do not carry a principal.

src/google/adk/a2a/converters/request_converter.py

@@ -25,6 +25,7 @@
 from pydantic import BaseModel
 
 from .. import _compat
+from ...agents._caller_principal import CallerPrincipal
 from ...agents.run_config import RunConfig
 from ..experimental import a2a_experimental
 from .part_converter import A2APartToGenAIPartConverter
@@ -45,6 +46,7 @@ class AgentRunRequest(BaseModel):
   new_message: Optional[genai_types.Content] = None
   state_delta: Optional[dict[str, Any]] = None
   run_config: Optional[RunConfig] = None
+  caller_principal: Optional[CallerPrincipal] = None
 
 
 A2ARequestToAgentRunRequestConverter = Callable[
@@ -86,6 +88,35 @@ def _get_user_id(request: RequestContext) -> str:
   return f'A2A_USER_{request.context_id}'
 
 
+def build_caller_principal(request: RequestContext) -> CallerPrincipal:
+  """Records what the A2A serving layer established about the caller.
+
+  ``_get_user_id`` above already reads ``call_context.user``, but it collapses
+  the result to a bare string and drops the one bit that matters downstream:
+  whether an authenticator produced that name or whether it was synthesized
+  from the caller-supplied context id. This keeps that bit.
+
+  Args:
+    request: The incoming request context from the A2A server.
+
+  Returns:
+    An authenticated principal when the A2A server authenticated the caller,
+    and an unauthenticated one otherwise. Never ``None``: an A2A request has
+    crossed a remote trust boundary either way, and ``None`` has to keep
+    meaning that no boundary was crossed at all.
+  """
+  user = request.call_context.user if request.call_context else None
+  user_name = getattr(user, 'user_name', None) if user is not None else None
+  # Authenticated exactly when _get_user_id above uses the name: a non-empty
+  # user_name on a principal that reports is_authenticated. The principal and
+  # the user id can therefore never disagree about the same request.
+  if not isinstance(user_name, str) or not user_name:
+    return CallerPrincipal(authenticated=False, source='a2a')
+  if not getattr(user, 'is_authenticated', False):
+    return CallerPrincipal(authenticated=False, source='a2a')
+  return CallerPrincipal(authenticated=True, user_name=user_name, source='a2a')
+
+
 @functools.lru_cache(maxsize=1)
 def _warn_unauthenticated_user_name_once() -> None:
   logger.warning(
@@ -137,4 +168,5 @@ def convert_a2a_request_to_agent_run_request(
           parts=output_parts,
       ),
       run_config=RunConfig(custom_metadata=custom_metadata),
+      caller_principal=build_caller_principal(request),
   )

src/google/adk/a2a/executor/a2a_agent_executor.py

@@ -34,6 +34,7 @@
 from ...utils.context_utils import Aclosing
 from ..agent.interceptors.new_integration_extension import _NEW_A2A_ADK_INTEGRATION_EXTENSION
 from ..converters.request_converter import AgentRunRequest
+from ..converters.request_converter import build_caller_principal
 from ..converters.utils import _get_adk_metadata_key
 from ..experimental import a2a_experimental
 from .a2a_agent_executor_impl import _A2aAgentExecutor as ExecutorImpl
@@ -189,6 +190,11 @@ async def _handle_request(
         context,
         self._config.a2a_part_converter,
     )
+    # The request converter is pluggable, so the principal is re-derived here
+    # from the RequestContext the A2A server handed us. A custom converter must
+    # not be able to drop the principal, and must not be able to assert one
+    # the server never established.
+    run_request.caller_principal = build_caller_principal(context)
 
     # ensure the session exists
     session = await self._prepare_session(context, run_request, runner)
@@ -198,6 +204,7 @@ async def _handle_request(
         session=session,
         new_message=run_request.new_message,
         run_config=run_request.run_config,
+        caller_principal=run_request.caller_principal,
     )
 
     executor_context = ExecutorContext(

src/google/adk/a2a/executor/a2a_agent_executor_impl.py

@@ -37,6 +37,7 @@
 from ..converters.long_running_functions import handle_user_input
 from ..converters.long_running_functions import LongRunningFunctions
 from ..converters.request_converter import AgentRunRequest
+from ..converters.request_converter import build_caller_principal
 from ..converters.utils import _get_adk_metadata_key
 from ..experimental import a2a_experimental
 from .config import A2aAgentExecutorConfig
@@ -102,6 +103,11 @@ async def execute(
           context,
           self._config.a2a_part_converter,
       )
+      # The request converter is pluggable, so the principal is re-derived
+      # here from the RequestContext the A2A server handed us. A custom
+      # converter must not be able to drop the principal, and must not be
+      # able to assert one the server never established.
+      run_request.caller_principal = build_caller_principal(context)
       session_id = await self._resolve_session(run_request, runner)
 
       executor_context = ExecutorContext(

src/google/adk/agents/__init__.py

@@ -42,6 +42,7 @@
     'Agent': '.llm_agent',
     'BaseAgent': '.base_agent',
     'BaseAgentConfig': '.base_agent_config',
+    'CallerPrincipal': '._caller_principal',
     'Context': '.context',
     'InvocationContext': '.invocation_context',
     'LiveRequest': '.live_request_queue',
@@ -68,6 +69,7 @@
     'McpInstructionProvider',
     'ParallelAgent',
     'SequentialAgent',
+    'CallerPrincipal',
     'InvocationContext',
     'LiveRequest',
     'LiveRequestQueue',