@@ -70,6 +70,7 @@
 from typing_extensions import override
 import yaml
 
+from ..agents._caller_principal import CallerPrincipal
 from ..agents._streaming_mode import StreamingMode
 from ..agents.base_agent import BaseAgent
 from ..agents.llm_agent import LlmAgent
@@ -447,23 +448,68 @@ def _accepts_kwargs(func: Callable[..., Any], kwargs: dict[str, Any]) -> bool:
   return True
 
 
+def _run_async_accepts(runner: Runner, name: str) -> bool:
+  """Whether runner.run_async takes the keyword ``name`` (or ``**kwargs``)."""
+  try:
+    params = inspect.signature(runner.run_async).parameters
+  except (ValueError, TypeError):
+    return True
+  return name in params or any(
+      p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values()
+  )
+
+
 def _with_abort_signal_kwarg(
     runner: Runner,
     kwargs: dict[str, Any],
     abort_signal: asyncio.Event,
 ) -> dict[str, Any]:
   """Adds abort_signal to kwargs when runner.run_async accepts it."""
-  try:
-    params = inspect.signature(runner.run_async).parameters
-    if "abort_signal" in params or any(
-        p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values()
-    ):
-      kwargs["abort_signal"] = abort_signal
-  except (ValueError, TypeError):
+  if _run_async_accepts(runner, "abort_signal"):
     kwargs["abort_signal"] = abort_signal
   return kwargs
 
 
+def _with_caller_principal_kwarg(
+    runner: Runner,
+    kwargs: dict[str, Any],
+    caller_principal: CallerPrincipal,
+) -> dict[str, Any]:
+  """Adds caller_principal to kwargs when runner.run_async accepts it."""
+  if _run_async_accepts(runner, "caller_principal"):
+    kwargs["caller_principal"] = caller_principal
+  return kwargs
+
+
+_API_SERVER_PRINCIPAL_SOURCE = "api_server"
+
+
+def _caller_principal_for_request(
+    request: Optional[Request],
+) -> CallerPrincipal:
+  """Returns the caller identity to run a /run or /run_sse request as.
+
+  The API server does not authenticate callers itself. A deployment that does,
+  in middleware or a dependency in front of these routes, states the result on
+  ``request.state.caller_principal`` and that value is used as is. Without
+  one, the request is recorded as unauthenticated from the ``api_server``
+  edge, so a tool confirmation it carries is warned about (default) or refused
+  (strict mode), the same decision made for an unauthenticated A2A caller.
+  The ``user_id`` in the request body is a claim the caller made rather than
+  one the server checked, so it is not consulted here.
+  """
+  principal = (
+      getattr(request.state, "caller_principal", None)
+      if request is not None
+      else None
+  )
+  if isinstance(principal, CallerPrincipal):
+    return principal
+  return CallerPrincipal(
+      authenticated=False, source=_API_SERVER_PRINCIPAL_SOURCE
+  )
+
+
 _current_session_options: contextvars.ContextVar[Optional[dict[str, Any]]] = (
     contextvars.ContextVar("current_session_options", default=None)
 )
@@ -2123,6 +2169,7 @@ async def run_agent(req: RunAgentRequest, request: Request) -> Response:
         )
 
       abort_signal = asyncio.Event()
+      caller_principal = _caller_principal_for_request(request)
 
       async def worker() -> list[Event]:
         run_async_kwargs = _with_abort_signal_kwarg(
@@ -2137,6 +2184,9 @@ async def worker() -> list[Event]:
             },
             abort_signal,
         )
+        run_async_kwargs = _with_caller_principal_kwarg(
+            runner, run_async_kwargs, caller_principal
+        )
         try:
           async with Aclosing(runner.run_async(**run_async_kwargs)) as agen:
             return [public_event(event) async for event in agen]
@@ -2186,7 +2236,9 @@ async def monitor() -> None:
         monitor_task.cancel()
 
     @app.post("/run_sse")
-    async def run_agent_sse(req: RunAgentRequest) -> StreamingResponse:
+    async def run_agent_sse(
+        req: RunAgentRequest, request: Request = None
+    ) -> StreamingResponse:
       app_name = req.app_name or self.default_app_name
       if not app_name:
         raise HTTPException(
@@ -2235,6 +2287,8 @@ async def run_agent_sse(req: RunAgentRequest) -> StreamingResponse:
               detail=f"Session not found: {req.session_id}",
           )
 
+      caller_principal = _caller_principal_for_request(request)
+
       # Convert the events to properly formatted SSE
       async def event_generator() -> AsyncIterator[str]:
         is_closing = False
@@ -2257,6 +2311,9 @@ async def _produce_events() -> None:
               },
               abort_signal,
           )
+          run_async_kwargs = _with_caller_principal_kwarg(
+              runner, run_async_kwargs, caller_principal
+          )
           try:
             async with Aclosing(runner.run_async(**run_async_kwargs)) as agen:
               try: