[BREAKING] Python: Require explicit AG-UI authorization scope

docs/specs/004-python-function-calling-loop.md

@@ -501,6 +501,10 @@ that manually replay messages own the equivalent rule: do not resend an approval
   observable warning/log. A nested `call_id` is never accepted as an occurrence-identity alias.
 - Session-backed pending snapshots are trusted host state and require tenant-scoped, authorized storage. Consume-on-bind
   does not claim durable exactly-once behavior across crashes or concurrent workers.
+- HTTP adapters that retain approval authority across requests must require an authenticated authorization scope or an
+  explicit single-user declaration. Client-owned conversation or thread identifiers never establish approval
+  ownership. A configured scope resolver that fails or returns an invalid value must fail before accessing pending
+  approval state and must not fall back to unscoped operation.
 - Tool lookup uses the recorded name against the current registry. A same-name implementation upgrade is allowed;
   removing the name prevents local execution.
 - Only the strict boolean `True` grants approval. Missing decisions and non-boolean values are rejection, not consent.

python/packages/ag-ui/AGENTS.md

@@ -17,6 +17,7 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard.
 ## Types
 
 - **`AGUIRequest`** / **`AGUIChatOptions`** - Request types
+- **`AuthorizationScopeResolver`** - FastAPI dependency callable that resolves authenticated endpoint ownership
 - **`AGUIThreadSnapshot`** / **`AGUIThreadSnapshotStore`** - Thread snapshot model with client-replayable data,
   private Session Continuation State, and a scoped async store protocol
 - **`availableInterrupts` / `resume`** - Optional canonical AG-UI `Interrupt` and `ResumeEntry` protocol data
@@ -120,8 +121,13 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard.
   in different trusted scopes cannot collide in context-provider state. The deprecated
   `legacy_session_id_from_thread_id=True` compatibility option preserves raw provider keys only for explicit
   migrations and must warn because it disables that isolation; it is unsafe for shared multi-tenant deployments.
-  A configured endpoint resolver must return a non-empty string. Reject invalid results before accessing state or
-  invoking the runner; absence of a resolver, not an invalid result, selects intentionally unscoped operation.
+  Endpoint registration must explicitly provide `authorization_scope_resolver`, retain the compatible
+  `snapshot_scope_resolver`, or declare `single_user=True`. The authorization resolver is a FastAPI dependency and
+  must return a non-empty string derived from authenticated request context. Reject invalid results before accessing
+  state or invoking the runner; never fall back to single-user or unscoped operation after a configured resolver fails.
+  `single_user=True` assigns one opaque authorization scope for the lifetime of that endpoint registration and must not
+  be used for callers with distinct authorization. Use an explicit resolver when snapshots, checkpoints, or provider
+  state must retain the same scope across process restarts.
 - Request `state` is client-owned Shared State and is merged into `AgentSession.state`, minus a protected set:
   tool-approval state, history/context-provider namespaces, message-injection state, standalone FIDES security
   state, and provider-owned keys.
@@ -192,7 +198,7 @@ from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint
 from fastapi import FastAPI
 
 app = FastAPI()
-add_agent_framework_fastapi_endpoint(app, agent)
+add_agent_framework_fastapi_endpoint(app, agent, single_user=True)
 ```
 
 ## Import Path

python/packages/ag-ui/README.md

@@ -39,7 +39,8 @@ agent = Agent(
 
 # Create FastAPI app and add AG-UI endpoint
 app = FastAPI()
-add_agent_framework_fastapi_endpoint(app, agent, "/")
+# Local/single-user only. Shared services must provide authorization_scope_resolver.
+add_agent_framework_fastapi_endpoint(app, agent, "/", single_user=True)
 
 # Run with: uvicorn main:app --reload
 ```
@@ -60,7 +61,8 @@ async def start(message: str, ctx: WorkflowContext) -> None:
 workflow = WorkflowBuilder(start_executor=start).build()
 
 app = FastAPI()
-add_agent_framework_fastapi_endpoint(app, workflow, "/")
+# Local/single-user only. Shared services must provide authorization_scope_resolver.
+add_agent_framework_fastapi_endpoint(app, workflow, "/", single_user=True)
 ```
 
 ### Server (Thread-Scoped WorkflowBuilder)
@@ -83,7 +85,8 @@ thread_scoped_workflow = AgentFrameworkWorkflow(
     workflow_factory=build_workflow_for_thread,
     name="my_workflow",
 )
-add_agent_framework_fastapi_endpoint(app, thread_scoped_workflow, "/")
+# Local/single-user only. Shared services must provide authorization_scope_resolver.
+add_agent_framework_fastapi_endpoint(app, thread_scoped_workflow, "/", single_user=True)
 ```
 
 ### Client (Connect to an AG-UI Server)
@@ -303,7 +306,7 @@ Review focus: whether these names are the right stable contract for Python users
 
 | Surface | Public exports |
 | --- | --- |
-| `agent_framework.ag_ui` facade | `AgentFrameworkAgent`, `AgentFrameworkWorkflow`, `AGUIChatClient`, `AGUIEventConverter`, `AGUIHttpService`, `AGUIThreadSnapshot`, `AGUIThreadSnapshotStore`, `InMemoryAGUIThreadSnapshotStore`, `SnapshotScopeResolver`, `add_agent_framework_fastapi_endpoint`, `state_carrier`, `state_update`, `__version__` |
+| `agent_framework.ag_ui` facade | `AgentFrameworkAgent`, `AgentFrameworkWorkflow`, `AGUIChatClient`, `AGUIEventConverter`, `AGUIHttpService`, `AGUIThreadSnapshot`, `AGUIThreadSnapshotStore`, `AuthorizationScopeResolver`, `InMemoryAGUIThreadSnapshotStore`, `SnapshotScopeResolver`, `add_agent_framework_fastapi_endpoint`, `state_carrier`, `state_update`, `__version__` |
 | Direct `agent_framework_ag_ui` package | Facade exports plus `AGUIChatOptions`, `AGUIRequest`, `AGUIThreadID`, `AgentState`, `DEFAULT_MAX_THREAD_SNAPSHOTS`, `DEFAULT_TAGS`, `PredictStateConfig`, `RunMetadata`, `SnapshotScope`, `WorkflowFactory` |
 | AG-UI protocol package (`ag_ui.core`) | `Interrupt`, `ResumeEntry`, `RunFinishedInterruptOutcome`, and related run outcome models |
 
@@ -332,13 +335,49 @@ Additional compatibility and draft support:
 
 ## Security: Authentication & Authorization
 
-The AG-UI endpoint does not enforce authentication by default. **For production deployments, you should add authentication** using FastAPI's dependency injection system via the `dependencies` parameter.
+Every AG-UI endpoint must choose an authorization mode:
+
+- **Authenticated/shared applications:** provide `authorization_scope_resolver`. FastAPI invokes it as a dependency, so
+  it can declare `Depends`, `Security`, header, cookie, or `Request` parameters. It must return a stable, non-empty
+  scope for the authenticated principal or collaboration boundary that may access the same server-side AG-UI state.
+- **Local or genuinely single-user applications:** set `single_user=True`. This deliberately treats every caller as
+  one authorization principal and assigns an opaque scope for the lifetime of that endpoint registration. Do not use
+  it on a shared service or where durable state must retain the same scope across process restarts.
+- **Compatibility:** existing `snapshot_scope_resolver` integrations remain supported and provide the same trusted
+  scope to snapshots, internal sessions, workflows, detached-run admission, and approvals.
+
+Authentication in the separate `dependencies` list can allow or reject a request, but its return value does not scope
+server-side state. Do not combine authentication-only dependencies with `single_user=True` when different authenticated
+callers can reach the endpoint.
+
+### Stable Single-User Scope
+
+`single_user=True` creates an opaque scope for one endpoint registration, which is appropriate for local and
+process-local examples. A single-user application that uses durable snapshots, checkpoints, or provider state can
+instead provide an application-owned constant through the normal resolver:
+
+```python
+def resolve_single_user_scope() -> str:
+    # Keep this stable for the lifetime of the persisted state and unique to this application boundary.
+    return "my-app:single-user:v1"
+
+
+add_agent_framework_fastapi_endpoint(
+    app,
+    agent,
+    "/",
+    authorization_scope_resolver=resolve_single_user_scope,
+)
+```
+
+The value is a server-owned namespace, not a credential. Do not read it from AG-UI request state or other
+client-controlled input.
 
 ### API Key Authentication Example
 
 ```python
 import os
-from fastapi import Depends, FastAPI, HTTPException, Security
+from fastapi import FastAPI, HTTPException, Security
 from fastapi.security import APIKeyHeader
 from agent_framework import Agent
 from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint
@@ -348,28 +387,29 @@ API_KEY_HEADER = APIKeyHeader(name="X-API-Key", auto_error=False)
 EXPECTED_API_KEY = os.environ.get("AG_UI_API_KEY")
 
 
-async def verify_api_key(api_key: str | None = Security(API_KEY_HEADER)) -> None:
-    """Verify the API key provided in the request header."""
+async def resolve_authorization_scope(api_key: str | None = Security(API_KEY_HEADER)) -> str:
+    """Verify the API key and return the state-ownership scope."""
     if not api_key or api_key != EXPECTED_API_KEY:
         raise HTTPException(status_code=401, detail="Invalid or missing API key")
+    return "configured-api-key"
 
 
 # Create agent and app
 agent = Agent(name="my_agent", instructions="...", client=...)
 app = FastAPI()
 
-# Register endpoint WITH authentication
+# Register endpoint with authentication-backed authorization scoping
 add_agent_framework_fastapi_endpoint(
     app,
     agent,
     "/",
-    dependencies=[Depends(verify_api_key)],  # Authentication enforced here
+    authorization_scope_resolver=resolve_authorization_scope,
 )
 ```
 
 ### Other Authentication Options
 
-The `dependencies` parameter accepts any FastAPI dependency, enabling integration with:
+The authorization resolver is a normal FastAPI dependency, enabling integration with:
 
 - **OAuth 2.0 / OpenID Connect** - Use `fastapi.security.OAuth2PasswordBearer`
 - **JWT Tokens** - Validate tokens with libraries like `python-jose`
@@ -418,21 +458,18 @@ agent = ...
 snapshot_store = InMemoryAGUIThreadSnapshotStore(max_snapshots=500)
 
 
-def resolve_snapshot_scope(request):
-    # Local demo scope. Production apps should derive the scope from authenticated user or tenant context.
-    del request
-    return "local-demo"
-
-
 add_agent_framework_fastapi_endpoint(
     app,
     agent,
     "/",
     snapshot_store=snapshot_store,
-    snapshot_scope_resolver=resolve_snapshot_scope,
+    single_user=True,
 )
 ```
 
+For a shared deployment, replace `single_user=True` with the authenticated
+`authorization_scope_resolver` shown above.
+
 Configured stateless agent snapshot stores are updated after function/MCP tool-result batches and approval safe points,
 then written once more with the terminal run state. These boundaries capture completed model/tool rounds without
 persisting sibling text or reasoning whose stream finalizer may still reject it. Intermediate snapshots project only
@@ -462,7 +499,7 @@ add_agent_framework_fastapi_endpoint(
     agent,
     "/",
     snapshot_store=snapshot_store,
-    snapshot_scope_resolver=resolve_snapshot_scope,
+    single_user=True,
     detached_runs=True,
     max_detached_runs=32,
     detached_run_timeout_seconds=3600,
@@ -486,14 +523,14 @@ FastAPI dependencies and other request-scoped resources may be released after th
 authorization, Snapshot Scope, and other durable values before the run starts; detached tools and providers must not
 retain request-owned clients or sessions that are expected to close with the HTTP request.
 
-Endpoint configuration requires `snapshot_scope_resolver` whenever a snapshot store is configured, including when
-the store is already set on a pre-wrapped `AgentFrameworkAgent` or `AgentFrameworkWorkflow`. The resolver returns
-the application-defined Snapshot Scope used with the AG-UI Thread id as the storage key. The endpoint also derives
-the internal `AgentSession.session_id` from this trusted scope and the client-owned Thread id, so context providers
-cannot merge server-side state for equal Thread ids in different scopes. The raw Thread id remains unchanged in
-AG-UI events and snapshot operations. When using `AgentFrameworkWorkflow(workflow_factory=...)`, the same resolver
-also scopes the in-memory workflow cache even without a snapshot store; provide it in multi-user deployments so two
-users who submit the same `threadId` do not share a live `Workflow` instance.
+Endpoint registration requires `authorization_scope_resolver`, the compatible `snapshot_scope_resolver`, or an
+explicit `single_user=True` declaration. The authenticated resolver returns the application-defined scope used with
+the AG-UI Thread id for snapshot storage, internal `AgentSession.session_id` derivation, workflow ownership,
+detached-run admission, and approval state. Context providers and workflow instances therefore cannot merge
+server-side state for equal Thread ids in different scopes. The raw Thread id remains unchanged in AG-UI events and
+snapshot operations. `single_user=True` assigns one opaque scope per endpoint registration, including endpoints that
+share the same route path or backing objects in one process. That scope changes when the endpoint is registered again;
+use an explicit resolver for durable snapshots, checkpoints, or provider state that must survive process restarts.
 
 A workflow request that omits `threadId` is assigned a generated Thread id. It is reported in `RUN_STARTED`, keys the
 `workflow_factory` instance and Thread Snapshot, and owns that run's pending `request_info` interrupts. Send it back as
@@ -505,9 +542,8 @@ created by such a run resumes with `checkpointId` and either the reported `threa
 Authenticate and authorize the current scope on every request, including hydration and approval resume, rather than
 trusting a scope supplied through Shared State or forwarded properties. A configured resolver must return a non-empty
 string. Returning `None`, an empty string, or another type fails the request with a generic HTTP 500 configuration
-error before accessing snapshots, approval state, or context providers. Resolver failures never fall back to unscoped
-operation. Valid scope strings are used exactly as returned, without trimming or normalization. An endpoint without a
-resolver remains intentionally unscoped; it must not share session-keyed storage across distinct authorization scopes.
+error before accessing snapshots, approval state, or context providers. Resolver failures never fall back to
+single-user or unscoped operation. Valid scope strings are used exactly as returned, without trimming or normalization.
 Use endpoint authentication dependencies to reject unauthorized requests before scope resolution.
 
 Existing applications that need time to migrate provider records from raw Thread-id keys can temporarily wrap the

python/packages/ag-ui/agent_framework_ag_ui/__init__.py

@@ -7,7 +7,7 @@
 
 from ._agent import AgentFrameworkAgent
 from ._client import AGUIChatClient
-from ._endpoint import add_agent_framework_fastapi_endpoint
+from ._endpoint import AuthorizationScopeResolver, add_agent_framework_fastapi_endpoint
 from ._event_converters import AGUIEventConverter
 from ._http_service import AGUIHttpService
 from ._message_adapters import agent_framework_messages_to_agui_host_history
@@ -38,6 +38,7 @@
 __all__ = [
     "AgentFrameworkAgent",
     "AgentFrameworkWorkflow",
+    "AuthorizationScopeResolver",
     "WorkflowFactory",
     "add_agent_framework_fastapi_endpoint",
     "agent_framework_messages_to_agui_host_history",