From 0594b266b8b534277791c8a69f71e068267a4241 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 15 Oct 2026 14:07:17 +0200 Subject: [PATCH 1/4] [BREAKING] Python: Require explicit AG-UI authorization scope --- .../specs/004-python-function-calling-loop.md | 4 + python/packages/ag-ui/AGENTS.md | 11 +- python/packages/ag-ui/README.md | 71 +++--- .../ag-ui/agent_framework_ag_ui/__init__.py | 3 +- .../ag-ui/agent_framework_ag_ui/_endpoint.py | 112 +++++++-- .../server/api/backend_tool_rendering.py | 3 + .../server/main.py | 13 +- .../packages/ag-ui/getting_started/server.py | 17 +- python/packages/ag-ui/tests/ag_ui/conftest.py | 1 + .../ag-ui/tests/ag_ui/test_endpoint.py | 213 +++++++++++++++--- .../ag-ui/tests/ag_ui/test_handoff_replay.py | 8 +- .../ag-ui/tests/ag_ui/test_http_round_trip.py | 4 +- .../ag-ui/tests/ag_ui/test_multi_turn.py | 2 +- .../ag-ui/tests/ag_ui/test_public_exports.py | 1 + .../core/agent_framework/ag_ui/__init__.py | 2 + .../ag_ui_workflow_handoff/backend/server.py | 3 + 16 files changed, 372 insertions(+), 96 deletions(-) diff --git a/docs/specs/004-python-function-calling-loop.md b/docs/specs/004-python-function-calling-loop.md index d5b9d1ec622..fe1a1c29428 100644 --- a/docs/specs/004-python-function-calling-loop.md +++ b/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. diff --git a/python/packages/ag-ui/AGENTS.md b/python/packages/ag-ui/AGENTS.md index 0ceb81e2e35..e0b6c318c55 100644 --- a/python/packages/ag-ui/AGENTS.md +++ b/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,12 @@ 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` intentionally preserves unscoped operation without snapshot persistence and must not be used for + callers with distinct authorization. - 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 +197,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 diff --git a/python/packages/ag-ui/README.md b/python/packages/ag-ui/README.md index 0e42d416d0e..6ce9ff289f6 100644 --- a/python/packages/ag-ui/README.md +++ b/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,25 @@ 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. Do not use it on a shared service. +- **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. ### 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 +363,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 +434,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 +475,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 +499,13 @@ 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` intentionally preserves unscoped behavior when no snapshot store is present; +when snapshot persistence is configured, the endpoint supplies a stable endpoint-local scope for the single user. 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 +517,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 diff --git a/python/packages/ag-ui/agent_framework_ag_ui/__init__.py b/python/packages/ag-ui/agent_framework_ag_ui/__init__.py index 74dc19935a2..1f87597703b 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/__init__.py +++ b/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", diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py b/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py index 18afc64f923..fa76888a674 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py @@ -7,15 +7,15 @@ import asyncio import copy import logging -from collections.abc import AsyncGenerator, Sequence +from collections.abc import AsyncGenerator, Awaitable, Callable, Sequence from contextlib import suppress from inspect import isawaitable -from typing import Any, cast +from typing import Any, TypeAlias, cast from ag_ui.core import RunErrorEvent from ag_ui.encoder import EventEncoder from agent_framework import CheckpointStorage, SupportsAgentRun, Workflow -from fastapi import FastAPI, HTTPException +from fastapi import FastAPI, HTTPException, Request from fastapi.params import Depends from fastapi.responses import JSONResponse, Response, StreamingResponse @@ -33,9 +33,14 @@ logger = logging.getLogger(__name__) +AuthorizationScopeResolver: TypeAlias = Callable[..., str | Awaitable[str]] +"""FastAPI dependency callable that returns the authenticated AG-UI authorization scope.""" + _DETACHED_READER_START_TIMEOUT_SECONDS = 1.0 _DETACHED_STREAM_QUEUE_SIZE = 16 _KEEPALIVE_COMMENT = "keepalive" +_AUTHORIZATION_SCOPE_STATE_KEY = "_agent_framework_ag_ui_authorization_scope" +_SINGLE_USER_SNAPSHOT_SCOPE_PREFIX = "ag-ui:single-user:" def _get_snapshot_store( @@ -60,7 +65,6 @@ def _configure_snapshot_persistence( protocol_runner: AgentFrameworkAgent | AgentFrameworkWorkflow, *, snapshot_store: AGUIThreadSnapshotStore | None, - snapshot_scope_resolver: SnapshotScopeResolver | None, ) -> None: existing_snapshot_store = _get_snapshot_store(protocol_runner) if snapshot_store is not None: @@ -68,14 +72,6 @@ def _configure_snapshot_persistence( raise ValueError("snapshot_store is already configured on the AG-UI runner.") if existing_snapshot_store is None: _set_snapshot_store(protocol_runner, snapshot_store) - existing_snapshot_store = snapshot_store - - if existing_snapshot_store is not None and snapshot_scope_resolver is None: - raise ValueError( - "snapshot_scope_resolver is required when snapshot_store is configured. " - "AG-UI Thread ids identify threads but do not authorize snapshot access; " - "provide a resolver that returns an explicit Snapshot Scope." - ) def _validate_keepalive_seconds(keepalive_seconds: float | None) -> None: @@ -90,6 +86,39 @@ def _validate_detached_run_options(max_detached_runs: int, detached_run_timeout_ raise ValueError("detached_run_timeout_seconds must be positive.") +def _validate_authorization_scope_configuration( + *, + authorization_scope_resolver: AuthorizationScopeResolver | None, + snapshot_scope_resolver: SnapshotScopeResolver | None, + single_user: bool, +) -> None: + if not isinstance(single_user, bool): + raise TypeError("single_user must be a bool.") + configured_modes = sum( + ( + authorization_scope_resolver is not None, + snapshot_scope_resolver is not None, + single_user, + ) + ) + if configured_modes == 0: + raise ValueError( + "AG-UI endpoint authorization must be explicit. " + "Provide authorization_scope_resolver for authenticated applications, " + "reuse snapshot_scope_resolver for compatibility, or declare single_user=True." + ) + if configured_modes > 1: + raise ValueError( + "Configure exactly one AG-UI authorization mode: authorization_scope_resolver, " + "snapshot_scope_resolver, or single_user=True." + ) + + +def _single_user_snapshot_scope(path: str) -> str: + """Return a stable endpoint-local scope for single-user snapshot persistence.""" + return f"{_SINGLE_USER_SNAPSHOT_SCOPE_PREFIX}{path}" + + def add_agent_framework_fastapi_endpoint( app: FastAPI, agent: SupportsAgentRun | AgentFrameworkAgent | Workflow | AgentFrameworkWorkflow, @@ -108,6 +137,8 @@ def add_agent_framework_fastapi_endpoint( detached_runs: bool = False, max_detached_runs: int = 32, detached_run_timeout_seconds: float = 3600, + authorization_scope_resolver: AuthorizationScopeResolver | None = None, + single_user: bool = False, ) -> None: """Add an AG-UI endpoint to a FastAPI app. @@ -123,15 +154,16 @@ def add_agent_framework_fastapi_endpoint( tags: OpenAPI tags for endpoint categorization (defaults to ["AG-UI"]) dependencies: Optional FastAPI dependencies for authentication/authorization. These dependencies run before the endpoint handler. Use this to add - authentication checks, rate limiting, or other middleware-like behavior. + authentication checks, rate limiting, or other middleware-like behavior. Their return values do not + establish the AG-UI authorization scope; use authorization_scope_resolver for that. Example: `dependencies=[Depends(verify_api_key)]` - snapshot_store: Optional AG-UI Thread Snapshot store. Snapshot persistence is opt-in and requires an - explicit Snapshot Scope resolver. - snapshot_scope_resolver: Optional resolver for the application-defined Snapshot Scope. Required whenever - a snapshot store is configured because an AG-UI Thread id is not an authorization boundary. Also scopes - the internal Agent Session id used by context providers and in-memory workflow_factory instances when - provided without a snapshot store. A configured resolver must return a non-empty string derived from - authorized request context; invalid results fail the request before accessing state or invoking the runner. + snapshot_store: Optional AG-UI Thread Snapshot store. Snapshot persistence is opt-in and requires either + an authorization resolver, the compatible Snapshot Scope resolver, or explicit single-user mode. + snapshot_scope_resolver: Compatible request-body resolver for the application-defined Snapshot Scope. Prefer + authorization_scope_resolver for new FastAPI applications because it can consume authenticated dependency + results. This resolver also scopes internal Agent Sessions, in-memory workflow_factory instances, and + approval state. It must return a non-empty string derived from authorized request context; invalid results + fail the request before accessing state or invoking the runner. checkpoint_storage: Optional workflow checkpoint storage, applied when the endpoint exposes a workflow. When provided, each run creates a checkpoint at the end of every superstep, and a run may resume from a persisted checkpoint by supplying its id in the AG-UI forwarded props @@ -152,6 +184,13 @@ def add_agent_framework_fastapi_endpoint( Additional requests receive HTTP 503 until capacity is released. detached_run_timeout_seconds: Maximum time a producer may remain active after its SSE reader disconnects or never starts. Defaults to 3600 seconds. Expired producers are cancelled and release endpoint capacity. + authorization_scope_resolver: Optional FastAPI dependency callable that returns a non-empty authorization + scope derived from authenticated request context. The callable may declare normal FastAPI dependencies, + headers, security schemes, or the current Request. Its result scopes snapshots, internal Agent Sessions, + workflow instances and checkpoints, detached-run admission, and server-side approval state. + single_user: Explicitly declare that every caller of this endpoint belongs to one authorization principal. + This preserves intentionally unscoped behavior for local development and single-user applications. + Never enable it on an endpoint shared by callers with distinct authorization. """ _validate_keepalive_seconds(keepalive_seconds) _validate_detached_run_options(max_detached_runs, detached_run_timeout_seconds) @@ -174,6 +213,12 @@ def add_agent_framework_fastapi_endpoint( else: raise TypeError("agent must be SupportsAgentRun, Workflow, AgentFrameworkAgent, or AgentFrameworkWorkflow.") + _validate_authorization_scope_configuration( + authorization_scope_resolver=authorization_scope_resolver, + snapshot_scope_resolver=snapshot_scope_resolver, + single_user=single_user, + ) + if checkpoint_storage is not None: if not isinstance(protocol_runner, AgentFrameworkWorkflow): raise ValueError("checkpoint_storage is only supported when the endpoint exposes a workflow.") @@ -189,12 +234,25 @@ def add_agent_framework_fastapi_endpoint( _configure_snapshot_persistence( protocol_runner, snapshot_store=snapshot_store, - snapshot_scope_resolver=snapshot_scope_resolver, ) background_tasks: set[asyncio.Task[Any]] = set() producer_tasks: set[asyncio.Task[None]] = set() active_runs: dict[tuple[str | None, str], asyncio.Task[None]] = {} + route_dependencies = list(dependencies or ()) + + if authorization_scope_resolver is not None: + + async def capture_authorization_scope( + request: Request, + resolved_scope: Any = Depends(authorization_scope_resolver), + ) -> None: + if not isinstance(resolved_scope, str) or not resolved_scope: + logger.error("authorization_scope_resolver must return a non-empty string.") + raise HTTPException(status_code=500, detail="An internal error has occurred.") + setattr(request.state, _AUTHORIZATION_SCOPE_STATE_KEY, resolved_scope) + + route_dependencies.append(Depends(capture_authorization_scope)) def retain_background_task(task: asyncio.Task[Any]) -> None: background_tasks.add(task) @@ -213,8 +271,8 @@ def task_done(completed: asyncio.Task[Any]) -> None: task.add_done_callback(task_done) - @app.post(path, tags=tags or ["AG-UI"], dependencies=dependencies, response_model=None) # type: ignore[arg-type] - async def agent_endpoint(request_body: AGUIRequest) -> Response: + @app.post(path, tags=tags or ["AG-UI"], dependencies=route_dependencies, response_model=None) # type: ignore[arg-type] + async def agent_endpoint(request_body: AGUIRequest, request: Request) -> Response: """Handle AG-UI agent requests. Note: Function is accessed via FastAPI's decorator registration, @@ -224,13 +282,19 @@ async def agent_endpoint(request_body: AGUIRequest) -> Response: input_data = request_body.model_dump(exclude_none=True) snapshot_persistence_active = _get_snapshot_store(protocol_runner) is not None snapshot_scope: str | None = None - if snapshot_scope_resolver is not None: + if authorization_scope_resolver is not None: + snapshot_scope = cast(str, getattr(request.state, _AUTHORIZATION_SCOPE_STATE_KEY)) + elif snapshot_scope_resolver is not None: resolved_scope = snapshot_scope_resolver(request_body) if isawaitable(resolved_scope): resolved_scope = await resolved_scope if not isinstance(resolved_scope, str) or not resolved_scope: raise ValueError("snapshot_scope_resolver must return a non-empty string.") snapshot_scope = resolved_scope + elif single_user and snapshot_persistence_active: + snapshot_scope = _single_user_snapshot_scope(path) + + if snapshot_scope is not None: input_data[_APPROVAL_SCOPE_INPUT_KEY] = snapshot_scope input_data[_SNAPSHOT_SCOPE_INPUT_KEY] = snapshot_scope if default_state: diff --git a/python/packages/ag-ui/agent_framework_ag_ui_examples/server/api/backend_tool_rendering.py b/python/packages/ag-ui/agent_framework_ag_ui_examples/server/api/backend_tool_rendering.py index 7db6ff5278a..388b9864cce 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui_examples/server/api/backend_tool_rendering.py +++ b/python/packages/ag-ui/agent_framework_ag_ui_examples/server/api/backend_tool_rendering.py @@ -25,4 +25,7 @@ def register_backend_tool_rendering(app: FastAPI) -> None: app, weather_agent(client), "/backend_tool_rendering", + # Local example only. For production, remove (or set False) and provide + # authorization_scope_resolver backed by authenticated request context. + single_user=True, ) diff --git a/python/packages/ag-ui/agent_framework_ag_ui_examples/server/main.py b/python/packages/ag-ui/agent_framework_ag_ui_examples/server/main.py index 82526f2ed0e..06e289d00d2 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui_examples/server/main.py +++ b/python/packages/ag-ui/agent_framework_ag_ui_examples/server/main.py @@ -12,7 +12,7 @@ import uvicorn from agent_framework import ChatOptions from agent_framework._clients import SupportsChatGetResponse -from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint +from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint as _add_agent_framework_fastapi_endpoint from agent_framework.azure import AzureOpenAIChatClient from agent_framework.openai import OpenAIChatCompletionClient from fastapi import FastAPI @@ -114,6 +114,17 @@ def _default_shared_client() -> SupportsChatGetResponse[ChatOptions]: client: SupportsChatGetResponse[ChatOptions] = _default_shared_client() + +def add_agent_framework_fastapi_endpoint(*args: Any, **kwargs: Any) -> None: + """Register local example endpoints under one explicit authorization principal. + + Production deployments must remove this default and pass an + authorization_scope_resolver backed by authenticated request context. + """ + kwargs.setdefault("single_user", True) + _add_agent_framework_fastapi_endpoint(*args, **kwargs) + + # Agentic Chat - basic chat agent add_agent_framework_fastapi_endpoint( app=app, diff --git a/python/packages/ag-ui/getting_started/server.py b/python/packages/ag-ui/getting_started/server.py index 4dfe0b36146..2e8988fa587 100644 --- a/python/packages/ag-ui/getting_started/server.py +++ b/python/packages/ag-ui/getting_started/server.py @@ -11,7 +11,7 @@ from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint from agent_framework.openai import OpenAIChatCompletionClient from dotenv import load_dotenv -from fastapi import Depends, FastAPI, HTTPException, Security +from fastapi import FastAPI, HTTPException, Security from fastapi.security import APIKeyHeader load_dotenv() @@ -56,12 +56,15 @@ 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 its stable authorization scope. Args: api_key: The API key from the X-API-Key header + Returns: + The authorization scope shared by requests using the configured API key. + Raises: HTTPException: If the API key is missing or invalid """ @@ -73,7 +76,7 @@ async def verify_api_key(api_key: str | None = Security(API_KEY_HEADER)) -> None "The endpoint is accessible without authentication. " "Set AG_UI_API_KEY to enable API key authentication." ) - return + return "local-single-user" if not api_key: raise HTTPException( @@ -86,6 +89,7 @@ async def verify_api_key(api_key: str | None = Security(API_KEY_HEADER)) -> None status_code=403, detail="Invalid API key.", ) + return "configured-api-key" # Server-side tool (executes on server) @@ -129,13 +133,12 @@ def get_time_zone(location: str) -> str: # Create FastAPI app app = FastAPI(title="AG-UI Server") -# Register the AG-UI endpoint with authentication -# The dependencies parameter accepts FastAPI Depends() objects that run before the handler +# Register the AG-UI endpoint with authentication-backed authorization scoping. add_agent_framework_fastapi_endpoint( app, agent, "/", - dependencies=[Depends(verify_api_key)], + authorization_scope_resolver=resolve_authorization_scope, ) if __name__ == "__main__": diff --git a/python/packages/ag-ui/tests/ag_ui/conftest.py b/python/packages/ag-ui/tests/ag_ui/conftest.py index c4d4d972cde..9e9367403bc 100644 --- a/python/packages/ag-ui/tests/ag_ui/conftest.py +++ b/python/packages/ag-ui/tests/ag_ui/conftest.py @@ -344,6 +344,7 @@ def _factory( state_schema=state_schema, predict_state_config=predict_state_config, default_state=default_state, + single_user=True, ) return app diff --git a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py index 7c8be9a9e83..fc0058369f3 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py +++ b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py @@ -62,7 +62,7 @@ SecureAgentConfig, ) from conftest import StubAgent # pyrefly: ignore[missing-import] # pyright: ignore[reportMissingImports] -from fastapi import FastAPI, Header, HTTPException +from fastapi import FastAPI, Header, HTTPException, Request from fastapi.params import Depends from fastapi.responses import StreamingResponse from fastapi.routing import APIRoute @@ -74,7 +74,9 @@ AGUIRequest, AGUIThreadSnapshot, InMemoryAGUIThreadSnapshotStore, - add_agent_framework_fastapi_endpoint, +) +from agent_framework_ag_ui import ( + add_agent_framework_fastapi_endpoint as _add_agent_framework_fastapi_endpoint, ) from agent_framework_ag_ui._agent import AgentFrameworkAgent from agent_framework_ag_ui._approval_lifecycle import ( @@ -91,6 +93,17 @@ ) +def add_agent_framework_fastapi_endpoint(*args: Any, **kwargs: Any) -> None: + """Register a test endpoint as intentionally single-user unless the test supplies a scope.""" + if ( + kwargs.get("authorization_scope_resolver") is None + and kwargs.get("snapshot_scope_resolver") is None + and not kwargs.get("single_user", False) + ): + kwargs["single_user"] = True + _add_agent_framework_fastapi_endpoint(*args, **kwargs) + + def _decode_sse_events(response: Any) -> list[dict[str, Any]]: content = response.content.decode("utf-8") return [json.loads(line[6:]) for line in content.splitlines() if line.startswith("data: ")] @@ -1331,21 +1344,21 @@ async def start(message: Any, ctx: WorkflowContext[Any, Any]) -> None: def test_add_endpoint_keepalive_default_is_enabled() -> None: """Keepalive defaults to the endpoint-owned enabled interval.""" - parameter = signature(add_agent_framework_fastapi_endpoint).parameters["keepalive_seconds"] + parameter = signature(_add_agent_framework_fastapi_endpoint).parameters["keepalive_seconds"] assert parameter.default == 15 def test_add_endpoint_detached_runs_default_is_disabled() -> None: """Detached execution is opt-in so current disconnect cancellation semantics remain the default.""" - parameter = signature(add_agent_framework_fastapi_endpoint).parameters["detached_runs"] + parameter = signature(_add_agent_framework_fastapi_endpoint).parameters["detached_runs"] assert parameter.default is False def test_add_endpoint_detached_run_limits_have_bounded_defaults() -> None: """Detached producers have finite admission and post-disconnect lifetime defaults.""" - parameters = signature(add_agent_framework_fastapi_endpoint).parameters + parameters = signature(_add_agent_framework_fastapi_endpoint).parameters assert parameters["max_detached_runs"].default == 32 assert parameters["detached_run_timeout_seconds"].default == 3600 @@ -1353,14 +1366,16 @@ def test_add_endpoint_detached_run_limits_have_bounded_defaults() -> None: def test_add_endpoint_detached_runs_preserves_existing_positional_parameter_order() -> None: """The new option is appended after existing parameters so positional a2ui_config callers do not shift.""" - parameters = list(signature(add_agent_framework_fastapi_endpoint).parameters) + parameters = list(signature(_add_agent_framework_fastapi_endpoint).parameters) assert parameters.index("detached_runs") > parameters.index("a2ui_config") + assert parameters.index("authorization_scope_resolver") > parameters.index("detached_run_timeout_seconds") + assert parameters.index("single_user") > parameters.index("authorization_scope_resolver") def test_add_endpoint_docstring_describes_keepalive_transport_behavior() -> None: """The public endpoint docs describe keepalive as transport comments, not AG-UI events.""" - docstring = add_agent_framework_fastapi_endpoint.__doc__ + docstring = _add_agent_framework_fastapi_endpoint.__doc__ assert docstring is not None normalized_docstring = " ".join(docstring.split()) @@ -1373,7 +1388,7 @@ def test_add_endpoint_docstring_describes_keepalive_transport_behavior() -> None def test_add_endpoint_docstring_describes_detached_run_behavior() -> None: """The public endpoint docs distinguish detached completion from resumable replay.""" - docstring = add_agent_framework_fastapi_endpoint.__doc__ + docstring = _add_agent_framework_fastapi_endpoint.__doc__ assert docstring is not None normalized_docstring = " ".join(docstring.split()) @@ -1652,7 +1667,8 @@ async def stream_fn( "threadId": "unstarted-thread", "messages": [{"role": "user", "content": "Start"}], } - ) + ), + Request({"type": "http", "method": "POST", "path": "/unstarted-reader", "headers": []}), ) assert isinstance(response, StreamingResponse) @@ -1714,7 +1730,8 @@ async def run(self, input_data: dict[str, Any]) -> AsyncGenerator[BaseEvent]: "threadId": "unstarted-thread", "messages": [{"role": "user", "content": "Start"}], } - ) + ), + Request({"type": "http", "method": "POST", "path": "/exact-capacity", "headers": []}), ) assert isinstance(response, StreamingResponse) await asyncio.wait_for(first_completed.wait(), timeout=5) @@ -8211,9 +8228,8 @@ async def test_endpoint_agent_approval_resume_wrong_thread_emits_run_error(): async def test_endpoint_agent_approval_resume_wrong_scope_emits_run_error_without_snapshot_store(): - """Approval State is scoped independently of AG-UI Thread Snapshots.""" + """Authenticated callers with the same Thread id cannot consume each other's pending approval.""" executed_cities: list[str] = [] - scope = {"value": "tenant-a"} def get_weather(city: str) -> str: executed_cities.append(city) @@ -8237,17 +8253,24 @@ def get_weather(city: str) -> str: updates=[AgentResponseUpdate(contents=[approval_request], role="assistant")], default_options={"tools": [weather_tool]}, ) + + async def resolve_authorization_scope(x_principal: str = Header()) -> str: + if x_principal not in {"tenant-a", "tenant-b"}: + raise HTTPException(status_code=403, detail="Unknown principal.") + return x_principal + app = FastAPI() add_agent_framework_fastapi_endpoint( app, AgentFrameworkAgent(agent=agent, require_confirmation=False), path="/approval-scoped", - snapshot_scope_resolver=lambda _request: scope["value"], + authorization_scope_resolver=resolve_authorization_scope, ) client = TestClient(app) pause_response = client.post( "/approval-scoped", + headers={"x-principal": "tenant-a"}, json={ "runId": "run-pause", "threadId": "thread-weather", @@ -8258,10 +8281,10 @@ def get_weather(city: str) -> str: pause_finished = [event for event in _decode_sse_events(pause_response) if event.get("type") == "RUN_FINISHED"] assert _run_finished_interrupts(pause_finished[-1])[0]["id"] == "call_get_weather" - scope["value"] = "tenant-b" agent.updates = [AgentResponseUpdate(contents=[Content.from_text(text="Should not run.")], role="assistant")] response = client.post( "/approval-scoped", + headers={"x-principal": "tenant-b"}, json={ "runId": "run-wrong-scope", "threadId": "thread-weather", @@ -8278,6 +8301,21 @@ def get_weather(city: str) -> str: assert executed_cities == [] assert not [event for event in events if event.get("type") == "TOOL_CALL_RESULT"] + owner_response = client.post( + "/approval-scoped", + headers={"x-principal": "tenant-a"}, + json={ + "runId": "run-owner-resume", + "threadId": "thread-weather", + "messages": [], + "resume": [{"interruptId": "call_get_weather", "status": "resolved", "payload": {"accepted": True}}], + }, + ) + + assert owner_response.status_code == 200 + assert executed_cities == ["Seattle"] + assert not [event for event in _decode_sse_events(owner_response) if event.get("type") == "RUN_ERROR"] + async def test_endpoint_agent_approval_function_name_mismatch_message_does_not_execute_tool(): """Client-supplied approval messages cannot swap the server-owned pending tool name.""" @@ -10817,12 +10855,11 @@ async def second_dependency(): async def test_endpoint_without_dependencies_is_accessible(build_chat_client): - """Test that endpoint without dependencies remains accessible (backward compatibility).""" + """An explicitly single-user endpoint remains accessible without authentication dependencies.""" app = FastAPI() agent = Agent(name="test", instructions="Test agent", client=build_chat_client()) - # No dependencies parameter - should be accessible without auth - add_agent_framework_fastapi_endpoint(app, agent, path="/open") + add_agent_framework_fastapi_endpoint(app, agent, path="/open", single_user=True) client = TestClient(app) response = client.post("/open", json={"messages": [{"role": "user", "content": "Hello"}]}) @@ -10839,24 +10876,124 @@ async def test_endpoint_invalid_agent_type_raises_typeerror(): add_agent_framework_fastapi_endpoint(app, agent="not_an_agent") # type: ignore[arg-type] # ty: ignore[invalid-argument-type] -async def test_endpoint_requires_snapshot_scope_resolver_when_store_configured(build_chat_client): - """Snapshot persistence setup must require an explicit Snapshot Scope resolver.""" +async def test_endpoint_requires_explicit_authorization_mode(build_chat_client): + """Endpoint registration must choose authenticated scoping or explicit single-user operation.""" + app = FastAPI() + agent = Agent(name="test", instructions="Test agent", client=build_chat_client()) + + with pytest.raises(ValueError, match="authorization must be explicit"): + _add_agent_framework_fastapi_endpoint(app, agent, path="/unconfigured") + + +async def test_endpoint_snapshot_store_requires_explicit_authorization_mode(build_chat_client): + """Snapshot persistence cannot bypass the endpoint's explicit authorization-mode requirement.""" app = FastAPI() agent = Agent(name="test", instructions="Test agent", client=build_chat_client()) store = InMemoryAGUIThreadSnapshotStore() - with pytest.raises(ValueError, match="snapshot_scope_resolver is required"): - add_agent_framework_fastapi_endpoint(app, agent, path="/snapshots", snapshot_store=store) + with pytest.raises(ValueError, match="authorization must be explicit"): + _add_agent_framework_fastapi_endpoint(app, agent, path="/snapshots", snapshot_store=store) + + +async def test_endpoint_wrapped_snapshot_store_requires_explicit_authorization_mode(build_chat_client): + """A pre-wrapped snapshot store cannot bypass the explicit authorization-mode requirement.""" + app = FastAPI() + runner = AgentFrameworkAgent( + agent=Agent(name="test", instructions="Test agent", client=build_chat_client()), + snapshot_store=InMemoryAGUIThreadSnapshotStore(), + ) + + with pytest.raises(ValueError, match="authorization must be explicit"): + _add_agent_framework_fastapi_endpoint(app, runner, path="/wrapped-snapshots") -async def test_endpoint_requires_snapshot_scope_resolver_when_wrapped_runner_has_store(build_chat_client): - """Pre-wrapped runners with snapshot stores must also provide a Snapshot Scope resolver.""" +async def test_endpoint_rejects_conflicting_authorization_modes(build_chat_client): + """An endpoint cannot silently choose between authenticated and single-user ownership.""" app = FastAPI() agent = Agent(name="test", instructions="Test agent", client=build_chat_client()) - wrapped_agent = AgentFrameworkAgent(agent=agent, snapshot_store=InMemoryAGUIThreadSnapshotStore()) - with pytest.raises(ValueError, match="snapshot_scope_resolver is required"): - add_agent_framework_fastapi_endpoint(app, wrapped_agent, path="/snapshots") + with pytest.raises(ValueError, match="exactly one AG-UI authorization mode"): + _add_agent_framework_fastapi_endpoint( + app, + agent, + path="/conflicting", + authorization_scope_resolver=lambda: "tenant-a", + single_user=True, + ) + + +async def test_endpoint_rejects_non_boolean_single_user_mode(build_chat_client): + """The explicit single-user escape hatch cannot be enabled by a truthy non-boolean value.""" + app = FastAPI() + agent = Agent(name="test", instructions="Test agent", client=build_chat_client()) + + with pytest.raises(TypeError, match="single_user must be a bool"): + _add_agent_framework_fastapi_endpoint( + app, + agent, + path="/invalid-single-user", + single_user="yes", # type: ignore[arg-type] + ) + + +async def test_endpoint_authorization_scope_resolver_uses_fastapi_dependencies(build_chat_client): + """The authorization resolver can consume normal FastAPI authentication dependencies.""" + app = FastAPI() + agent = Agent(name="test", instructions="Test agent", client=build_chat_client()) + authenticated: list[str] = [] + + async def authenticate(x_api_key: str | None = Header(None)) -> str: + if x_api_key != "secret-key": + raise HTTPException(status_code=401, detail="Unauthorized") + authenticated.append("principal-a") + return "principal-a" + + async def resolve_authorization_scope(principal: str = Depends(authenticate)) -> str: + return principal + + add_agent_framework_fastapi_endpoint( + app, + agent, + path="/scoped", + authorization_scope_resolver=resolve_authorization_scope, + ) + client = TestClient(app) + + unauthorized = client.post("/scoped", json={"messages": [{"role": "user", "content": "Hello"}]}) + authorized = client.post( + "/scoped", + headers={"x-api-key": "secret-key"}, + json={"messages": [{"role": "user", "content": "Hello"}]}, + ) + + assert unauthorized.status_code == 401 + assert authorized.status_code == 200 + assert authenticated == ["principal-a"] + + +async def test_endpoint_invalid_authorization_scope_fails_before_runner(build_chat_client): + """A configured resolver cannot fall back to unscoped operation.""" + app = FastAPI() + runner = AgentFrameworkAgent(agent=Agent(name="test", instructions="Test agent", client=build_chat_client())) + run_spy = AsyncMock(wraps=runner.run) + + async def invalid_scope() -> str: + return "" + + runner.run = run_spy # type: ignore[method-assign] + add_agent_framework_fastapi_endpoint( + app, + runner, + path="/invalid-scope", + authorization_scope_resolver=invalid_scope, + ) + client = TestClient(app, raise_server_exceptions=False) + + response = client.post("/invalid-scope", json={"messages": [{"role": "user", "content": "Hello"}]}) + + assert response.status_code == 500 + assert response.json() == {"detail": "An internal error has occurred."} + run_spy.assert_not_called() async def test_endpoint_accepts_snapshot_store_with_scope_resolver(build_chat_client): @@ -10883,6 +11020,30 @@ async def test_endpoint_accepts_snapshot_store_with_scope_resolver(build_chat_cl assert response.headers["content-type"] == "text/event-stream; charset=utf-8" +async def test_endpoint_accepts_snapshot_store_in_single_user_mode(build_chat_client): + """A single-user endpoint gets a stable internal scope when snapshot persistence needs one.""" + app = FastAPI() + agent = Agent(name="test", instructions="Test agent", client=build_chat_client()) + store = InMemoryAGUIThreadSnapshotStore() + + add_agent_framework_fastapi_endpoint( + app, + agent, + path="/single-user-snapshots", + snapshot_store=store, + single_user=True, + ) + client = TestClient(app) + + response = client.post( + "/single-user-snapshots", + json={"messages": [{"role": "user", "content": "Hello"}], "thread_id": "thread-1"}, + ) + + assert response.status_code == 200 + assert await store.get(scope="ag-ui:single-user:/single-user-snapshots", thread_id="thread-1") is not None + + @pytest.mark.parametrize("async_resolver", [False, True], ids=["sync", "async"]) @pytest.mark.parametrize("persist_snapshots", [False, True], ids=["stateless", "snapshots"]) @pytest.mark.parametrize("request_kind", ["run", "hydrate", "resume"]) diff --git a/python/packages/ag-ui/tests/ag_ui/test_handoff_replay.py b/python/packages/ag-ui/tests/ag_ui/test_handoff_replay.py index bd3b473ffd3..c273317f704 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_handoff_replay.py +++ b/python/packages/ag-ui/tests/ag_ui/test_handoff_replay.py @@ -135,7 +135,13 @@ async def test_outer_agent_replays_balanced_handoff_on_second_turn() -> None: workflow_factory=lambda _thread_id: _build_handoff_workflow(triage_requests) ) app = FastAPI() - add_agent_framework_fastapi_endpoint(app, workflow_runner, path="/workflow", keepalive_seconds=None) + add_agent_framework_fastapi_endpoint( + app, + workflow_runner, + path="/workflow", + keepalive_seconds=None, + single_user=True, + ) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as http_client: diff --git a/python/packages/ag-ui/tests/ag_ui/test_http_round_trip.py b/python/packages/ag-ui/tests/ag_ui/test_http_round_trip.py index 115bb46b5d6..20db812e435 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_http_round_trip.py +++ b/python/packages/ag-ui/tests/ag_ui/test_http_round_trip.py @@ -26,7 +26,7 @@ def _build_app_with_agent(updates: list[AgentResponseUpdate], **kwargs: Any) -> stub = StubAgent(updates=updates) agent = AgentFrameworkAgent(agent=stub, **kwargs) app = FastAPI() - add_agent_framework_fastapi_endpoint(app, agent) + add_agent_framework_fastapi_endpoint(app, agent, single_user=True) return app @@ -34,7 +34,7 @@ def _build_app_with_workflow(workflow_builder: WorkflowBuilder) -> FastAPI: workflow = workflow_builder.build() wrapper = AgentFrameworkWorkflow(workflow=workflow) app = FastAPI() - add_agent_framework_fastapi_endpoint(app, wrapper) + add_agent_framework_fastapi_endpoint(app, wrapper, single_user=True) return app diff --git a/python/packages/ag-ui/tests/ag_ui/test_multi_turn.py b/python/packages/ag-ui/tests/ag_ui/test_multi_turn.py index 3fa1cc5d245..06aa20f4272 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_multi_turn.py +++ b/python/packages/ag-ui/tests/ag_ui/test_multi_turn.py @@ -35,7 +35,7 @@ def _build_app_with_agent(updates: list[AgentResponseUpdate], **kwargs: Any) -> stub = StubAgent(updates=updates) agent = AgentFrameworkAgent(agent=stub, **kwargs) app = FastAPI() - add_agent_framework_fastapi_endpoint(app, agent) + add_agent_framework_fastapi_endpoint(app, agent, single_user=True) return app diff --git a/python/packages/ag-ui/tests/ag_ui/test_public_exports.py b/python/packages/ag-ui/tests/ag_ui/test_public_exports.py index 5e76248e888..8825852dbe8 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_public_exports.py +++ b/python/packages/ag-ui/tests/ag_ui/test_public_exports.py @@ -62,6 +62,7 @@ def test_core_ag_ui_lazy_exports_include_snapshot_primitives() -> None: """Core facade must expose snapshot primitives needed for endpoint configuration.""" from agent_framework import ag_ui + assert hasattr(ag_ui, "AuthorizationScopeResolver") assert hasattr(ag_ui, "AGUIThreadSnapshot") assert hasattr(ag_ui, "AGUIThreadSnapshotStore") assert hasattr(ag_ui, "InMemoryAGUIThreadSnapshotStore") diff --git a/python/packages/core/agent_framework/ag_ui/__init__.py b/python/packages/core/agent_framework/ag_ui/__init__.py index 3ec2375b247..fee50b301e9 100644 --- a/python/packages/core/agent_framework/ag_ui/__init__.py +++ b/python/packages/core/agent_framework/ag_ui/__init__.py @@ -13,6 +13,7 @@ - AGUIHttpService - AGUIThreadSnapshot - AGUIThreadSnapshotStore +- AuthorizationScopeResolver - InMemoryAGUIThreadSnapshotStore - SnapshotScopeResolver - add_agent_framework_fastapi_endpoint @@ -37,6 +38,7 @@ "AGUIHttpService", "AGUIThreadSnapshot", "AGUIThreadSnapshotStore", + "AuthorizationScopeResolver", "InMemoryAGUIThreadSnapshotStore", "SnapshotScopeResolver", "state_carrier", diff --git a/python/samples/05-end-to-end/ag_ui_workflow_handoff/backend/server.py b/python/samples/05-end-to-end/ag_ui_workflow_handoff/backend/server.py index c33774b65b6..08aaa65740d 100644 --- a/python/samples/05-end-to-end/ag_ui_workflow_handoff/backend/server.py +++ b/python/samples/05-end-to-end/ag_ui_workflow_handoff/backend/server.py @@ -315,6 +315,9 @@ def create_app() -> FastAPI: app=app, agent=demo_workflow, path="/handoff_demo", + # Local sample only. For production, remove (or set False) and provide + # authorization_scope_resolver backed by authenticated request context. + single_user=True, ) @app.get("/healthz") From 633cdfc2fedc4c3b48cabc1bc7929513a1e95984 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 15 Oct 2026 14:17:52 +0200 Subject: [PATCH 2/4] Python: Complete AG-UI authorization scope migration --- .../agent_framework_ag_ui_examples/README.md | 51 ++++++++++++++----- .../packages/ag-ui/getting_started/README.md | 6 ++- .../ag-ui/tests/ag_ui/test_endpoint.py | 6 +-- .../core/agent_framework/ag_ui/__init__.pyi | 2 + 4 files changed, 48 insertions(+), 17 deletions(-) diff --git a/python/packages/ag-ui/agent_framework_ag_ui_examples/README.md b/python/packages/ag-ui/agent_framework_ag_ui_examples/README.md index 69e85b3b5ea..e90ac2d0dc1 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui_examples/README.md +++ b/python/packages/ag-ui/agent_framework_ag_ui_examples/README.md @@ -23,13 +23,16 @@ from agent_framework_ag_ui_examples.agents import simple_agent, weather_agent app = FastAPI() +# These are local single-user examples. For production, replace single_user=True +# with an authorization_scope_resolver backed by authenticated request context. + # Option 1: Use Azure OpenAI azure_client = OpenAIChatCompletionClient(model="gpt-4") -add_agent_framework_fastapi_endpoint(app, simple_agent(azure_client), "/chat") +add_agent_framework_fastapi_endpoint(app, simple_agent(azure_client), "/chat", single_user=True) # Option 2: Use OpenAI openai_client = OpenAIChatClient(model="gpt-4o") -add_agent_framework_fastapi_endpoint(app, weather_agent(openai_client), "/weather") +add_agent_framework_fastapi_endpoint(app, weather_agent(openai_client), "/weather", single_user=True) # Run with: uvicorn main:app --reload ``` @@ -51,7 +54,9 @@ agent = Agent( # Create FastAPI app and add AG-UI endpoint app = FastAPI() -add_agent_framework_fastapi_endpoint(app, agent, "/agent") +# Local example only. For production, replace single_user=True with an +# authorization_scope_resolver backed by authenticated request context. +add_agent_framework_fastapi_endpoint(app, agent, "/agent", single_user=True) # Run with: uvicorn main:app --reload ``` @@ -179,15 +184,37 @@ app = FastAPI(title="AG-UI Examples") # Create a chat client (shared across all agents, or create individual ones) client = OpenAIChatCompletionClient(model="gpt-4") -# Add all example endpoints -add_agent_framework_fastapi_endpoint(app, simple_agent(client), "/agentic_chat") -add_agent_framework_fastapi_endpoint(app, weather_agent(client), "/backend_tool_rendering") -add_agent_framework_fastapi_endpoint(app, human_in_the_loop_agent(client), "/human_in_the_loop") -add_agent_framework_fastapi_endpoint(app, task_steps_agent_wrapped(client), "/agentic_generative_ui") # type: ignore[arg-type] -add_agent_framework_fastapi_endpoint(app, ui_generator_agent(client), "/tool_based_generative_ui") -add_agent_framework_fastapi_endpoint(app, recipe_agent(client), "/shared_state") -add_agent_framework_fastapi_endpoint(app, document_writer_agent(client), "/predictive_state_updates") -add_agent_framework_fastapi_endpoint(app, subgraphs_agent(), "/subgraphs") +# Add local single-user example endpoints. For production, replace +# single_user=True with an authorization_scope_resolver backed by authenticated +# request context. +add_agent_framework_fastapi_endpoint(app, simple_agent(client), "/agentic_chat", single_user=True) +add_agent_framework_fastapi_endpoint(app, weather_agent(client), "/backend_tool_rendering", single_user=True) +add_agent_framework_fastapi_endpoint( + app, + human_in_the_loop_agent(client), + "/human_in_the_loop", + single_user=True, +) +add_agent_framework_fastapi_endpoint( + app, + task_steps_agent_wrapped(client), + "/agentic_generative_ui", + single_user=True, +) # type: ignore[arg-type] +add_agent_framework_fastapi_endpoint( + app, + ui_generator_agent(client), + "/tool_based_generative_ui", + single_user=True, +) +add_agent_framework_fastapi_endpoint(app, recipe_agent(client), "/shared_state", single_user=True) +add_agent_framework_fastapi_endpoint( + app, + document_writer_agent(client), + "/predictive_state_updates", + single_user=True, +) +add_agent_framework_fastapi_endpoint(app, subgraphs_agent(), "/subgraphs", single_user=True) ``` ## Architecture diff --git a/python/packages/ag-ui/getting_started/README.md b/python/packages/ag-ui/getting_started/README.md index 142f002d6ec..aa8cd3e9e7c 100644 --- a/python/packages/ag-ui/getting_started/README.md +++ b/python/packages/ag-ui/getting_started/README.md @@ -241,8 +241,10 @@ agent = Agent( # Create FastAPI app app = FastAPI(title="AG-UI Server") -# Register the AG-UI endpoint -add_agent_framework_fastapi_endpoint(app, agent, "/") +# Register the local single-user AG-UI endpoint. +# For production, replace single_user=True with an authorization_scope_resolver +# backed by authenticated request context. +add_agent_framework_fastapi_endpoint(app, agent, "/", single_user=True) if __name__ == "__main__": import uvicorn diff --git a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py index fc0058369f3..202f568be1d 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py +++ b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py @@ -13,7 +13,7 @@ from dataclasses import dataclass from inspect import signature from pathlib import Path -from typing import Any, cast +from typing import Annotated, Any, cast from unittest.mock import AsyncMock, Mock import pytest @@ -10932,7 +10932,7 @@ async def test_endpoint_rejects_non_boolean_single_user_mode(build_chat_client): app, agent, path="/invalid-single-user", - single_user="yes", # type: ignore[arg-type] + single_user="yes", # type: ignore[arg-type] # ty: ignore[invalid-argument-type] ) @@ -10948,7 +10948,7 @@ async def authenticate(x_api_key: str | None = Header(None)) -> str: authenticated.append("principal-a") return "principal-a" - async def resolve_authorization_scope(principal: str = Depends(authenticate)) -> str: + async def resolve_authorization_scope(principal: Annotated[str, Depends(authenticate)]) -> str: return principal add_agent_framework_fastapi_endpoint( diff --git a/python/packages/core/agent_framework/ag_ui/__init__.pyi b/python/packages/core/agent_framework/ag_ui/__init__.pyi index 794aeae4a58..b51730743e5 100644 --- a/python/packages/core/agent_framework/ag_ui/__init__.pyi +++ b/python/packages/core/agent_framework/ag_ui/__init__.pyi @@ -8,6 +8,7 @@ from agent_framework_ag_ui import ( AGUIHttpService, AGUIThreadSnapshot, AGUIThreadSnapshotStore, + AuthorizationScopeResolver, InMemoryAGUIThreadSnapshotStore, SnapshotScopeResolver, __version__, @@ -25,6 +26,7 @@ __all__ = [ "AGUIThreadSnapshotStore", "AgentFrameworkAgent", "AgentFrameworkWorkflow", + "AuthorizationScopeResolver", "InMemoryAGUIThreadSnapshotStore", "SnapshotScopeResolver", "__version__", From 9b7e6f13a194889607f675b8a12063fb3d6d080a Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 15 Oct 2026 14:36:34 +0200 Subject: [PATCH 3/4] Python: Harden AG-UI registration scope resolution --- python/packages/ag-ui/AGENTS.md | 5 +- python/packages/ag-ui/README.md | 8 +- .../ag-ui/agent_framework_ag_ui/_endpoint.py | 22 ++- .../ag-ui/tests/ag_ui/test_endpoint.py | 143 +++++++++++++----- 4 files changed, 128 insertions(+), 50 deletions(-) diff --git a/python/packages/ag-ui/AGENTS.md b/python/packages/ag-ui/AGENTS.md index e0b6c318c55..bd79ecda92a 100644 --- a/python/packages/ag-ui/AGENTS.md +++ b/python/packages/ag-ui/AGENTS.md @@ -125,8 +125,9 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard. `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` intentionally preserves unscoped operation without snapshot persistence and must not be used for - callers with distinct authorization. + `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. diff --git a/python/packages/ag-ui/README.md b/python/packages/ag-ui/README.md index 6ce9ff289f6..146fff2937c 100644 --- a/python/packages/ag-ui/README.md +++ b/python/packages/ag-ui/README.md @@ -341,7 +341,8 @@ Every AG-UI endpoint must choose an authorization mode: 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. Do not use it on a shared service. + 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. @@ -504,8 +505,9 @@ explicit `single_user=True` declaration. The authenticated resolver returns the 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` intentionally preserves unscoped behavior when no snapshot store is present; -when snapshot persistence is configured, the endpoint supplies a stable endpoint-local scope for the single user. +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 diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py b/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py index fa76888a674..cb00ddf86c4 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py @@ -7,6 +7,7 @@ import asyncio import copy import logging +import uuid from collections.abc import AsyncGenerator, Awaitable, Callable, Sequence from contextlib import suppress from inspect import isawaitable @@ -40,7 +41,7 @@ _DETACHED_STREAM_QUEUE_SIZE = 16 _KEEPALIVE_COMMENT = "keepalive" _AUTHORIZATION_SCOPE_STATE_KEY = "_agent_framework_ag_ui_authorization_scope" -_SINGLE_USER_SNAPSHOT_SCOPE_PREFIX = "ag-ui:single-user:" +_SINGLE_USER_SCOPE_PREFIX = "ag-ui:single-user:" def _get_snapshot_store( @@ -114,9 +115,9 @@ def _validate_authorization_scope_configuration( ) -def _single_user_snapshot_scope(path: str) -> str: - """Return a stable endpoint-local scope for single-user snapshot persistence.""" - return f"{_SINGLE_USER_SNAPSHOT_SCOPE_PREFIX}{path}" +def _single_user_scope() -> str: + """Return one opaque scope that stays stable for an endpoint registration.""" + return f"{_SINGLE_USER_SCOPE_PREFIX}{uuid.uuid4().hex}" def add_agent_framework_fastapi_endpoint( @@ -189,8 +190,10 @@ def add_agent_framework_fastapi_endpoint( headers, security schemes, or the current Request. Its result scopes snapshots, internal Agent Sessions, workflow instances and checkpoints, detached-run admission, and server-side approval state. single_user: Explicitly declare that every caller of this endpoint belongs to one authorization principal. - This preserves intentionally unscoped behavior for local development and single-user applications. - Never enable it on an endpoint shared by callers with distinct authorization. + The endpoint receives one opaque scope that remains stable for that registration and separates equal route + paths registered by other applications. Use an explicit resolver when scope continuity must survive process + restarts or when storage is shared beyond one registration. Never enable this mode for callers with + distinct authorization. """ _validate_keepalive_seconds(keepalive_seconds) _validate_detached_run_options(max_detached_runs, detached_run_timeout_seconds) @@ -240,6 +243,7 @@ def add_agent_framework_fastapi_endpoint( producer_tasks: set[asyncio.Task[None]] = set() active_runs: dict[tuple[str | None, str], asyncio.Task[None]] = {} route_dependencies = list(dependencies or ()) + single_user_scope = _single_user_scope() if single_user else None if authorization_scope_resolver is not None: @@ -247,6 +251,8 @@ async def capture_authorization_scope( request: Request, resolved_scope: Any = Depends(authorization_scope_resolver), ) -> None: + if isawaitable(resolved_scope): + resolved_scope = await resolved_scope if not isinstance(resolved_scope, str) or not resolved_scope: logger.error("authorization_scope_resolver must return a non-empty string.") raise HTTPException(status_code=500, detail="An internal error has occurred.") @@ -291,8 +297,8 @@ async def agent_endpoint(request_body: AGUIRequest, request: Request) -> Respons if not isinstance(resolved_scope, str) or not resolved_scope: raise ValueError("snapshot_scope_resolver must return a non-empty string.") snapshot_scope = resolved_scope - elif single_user and snapshot_persistence_active: - snapshot_scope = _single_user_snapshot_scope(path) + elif single_user: + snapshot_scope = single_user_scope if snapshot_scope is not None: input_data[_APPROVAL_SCOPE_INPUT_KEY] = snapshot_scope diff --git a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py index 202f568be1d..d6cd10765d9 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py +++ b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py @@ -86,12 +86,15 @@ ResumeDecision, ) from agent_framework_ag_ui._approval_state import InMemoryAGUIApprovalStateStore, approval_state_thread_id +from agent_framework_ag_ui._snapshots import _session_id_for_thread from agent_framework_ag_ui._workflow import ( _CHECKPOINT_REQUEST_OWNER_KEY, AgentFrameworkWorkflow, _OwnedWorkflowCheckpointStorage, ) +_DEFAULT_AUTHORIZATION_SCOPE = "test-single-user" + def add_agent_framework_fastapi_endpoint(*args: Any, **kwargs: Any) -> None: """Register a test endpoint as intentionally single-user unless the test supplies a scope.""" @@ -100,7 +103,7 @@ def add_agent_framework_fastapi_endpoint(*args: Any, **kwargs: Any) -> None: and kwargs.get("snapshot_scope_resolver") is None and not kwargs.get("single_user", False) ): - kwargs["single_user"] = True + kwargs["snapshot_scope_resolver"] = lambda _request: _DEFAULT_AUTHORIZATION_SCOPE _add_agent_framework_fastapi_endpoint(*args, **kwargs) @@ -109,8 +112,12 @@ def _decode_sse_events(response: Any) -> list[dict[str, Any]]: return [json.loads(line[6:]) for line in content.splitlines() if line.startswith("data: ")] -def _unscoped_approval_thread_id(thread_id: str) -> str: - return approval_state_thread_id(scope=None, thread_id=thread_id) +def _default_approval_thread_id(thread_id: str) -> str: + return approval_state_thread_id(scope=_DEFAULT_AUTHORIZATION_SCOPE, thread_id=thread_id) + + +def _default_session_id(thread_id: str) -> str: + return _session_id_for_thread(scope=_DEFAULT_AUTHORIZATION_SCOPE, thread_id=thread_id) async def _post_until_sse_event_then_disconnect( @@ -3330,7 +3337,12 @@ async def authorize(x_user: str = Header(), x_scope: str = Header()) -> None: keepalive_seconds=None, ) add_agent_framework_fastapi_endpoint( - app, agent, path="/unscoped-memory", dependencies=[Depends(authorize)], keepalive_seconds=None + app, + agent, + path="/unscoped-memory", + dependencies=[Depends(authorize)], + keepalive_seconds=None, + single_user=True, ) client = TestClient(app) request_number = 0 @@ -3522,7 +3534,8 @@ def post( assert all(victim_description not in context for context in model_contexts[collision_start:]) assert {path: path.read_bytes() for path in victim_files} == victim_files - # No resolver is intentionally unscoped, unlike a configured resolver returning None. + # Explicit single-user mode gets one registration-local scope and retains continuity within that registration. + single_user_session_id: str | None = None for tool_name, arguments in [ ("file_memory_write", {"file_name": "unscoped.md", "content": "unscoped continuity"}), ("file_memory_read", {"file_name": "unscoped.md"}), @@ -3536,7 +3549,11 @@ def post( public_thread_id="ordinary-unscoped-thread", ) assert response.status_code == 200 - assert chat_client.last_session.session_id == "ordinary-unscoped-thread" + assert chat_client.last_session.session_id != "ordinary-unscoped-thread" + if single_user_session_id is None: + single_user_session_id = chat_client.last_session.session_id + else: + assert chat_client.last_session.session_id == single_user_session_id if tool_name == "file_memory_read": result = next( event["content"] for event in _decode_sse_events(response) if event["type"] == "TOOL_CALL_RESULT" @@ -4074,7 +4091,7 @@ async def stream_fn( assert [event["code"] for event in resume_events if event.get("type") == "RUN_ERROR"] == ["MiddlewareFailure"] assert not [event for event in resume_events if event.get("type") == "TOOL_CALL_RESULT"] assert executed == [] - assert policy_observations == [(f"thread-policy-{a2ui_mode}", None)] + assert policy_observations == [(_default_session_id(f"thread-policy-{a2ui_mode}"), None)] state["phase"] = "pause" direct_session = AgentSession(session_id=f"direct-policy-{a2ui_mode}") @@ -4221,7 +4238,7 @@ async def stream_fn( assert not [event for event in events if event.get("type") == "TOOL_CALL_RESULT"] assert executed == [] assert render_calls == [] - assert policy_sessions == [f"thread-mixed-policy-{a2ui_mode}"] + assert policy_sessions == [_default_session_id(f"thread-mixed-policy-{a2ui_mode}")] state["direct"] = True with pytest.raises(MiddlewareFailure): @@ -4598,7 +4615,7 @@ def second_tool() -> str: ) wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) lifecycle = wrapped_agent._approval_state_store.lifecycle - approval_thread_id = _unscoped_approval_thread_id("thread-shared-call") + approval_thread_id = _default_approval_thread_id("thread-shared-call") first = lifecycle.register( owner=ApprovalExecutionOwner.LOCAL, thread_id=approval_thread_id, @@ -4651,7 +4668,7 @@ async def test_endpoint_agent_unavailable_local_executor_releases_every_unstarte ) wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) lifecycle = wrapped_agent._approval_state_store.lifecycle - approval_thread_id = _unscoped_approval_thread_id("thread-unavailable-batch") + approval_thread_id = _default_approval_thread_id("thread-unavailable-batch") local = lifecycle.register( owner=ApprovalExecutionOwner.LOCAL, thread_id=approval_thread_id, @@ -5658,7 +5675,7 @@ def guarded_tool(value: str) -> str: wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) wrapped_agent._approval_state_store.lifecycle.register( owner=ApprovalExecutionOwner.LOCAL, - thread_id=_unscoped_approval_thread_id("thread-reused-confirm-id"), + thread_id=_default_approval_thread_id("thread-reused-confirm-id"), interrupt_id="af-call-current", call_id="provider-reused", name="guarded_tool", @@ -5729,7 +5746,7 @@ def second_tool() -> str: ) wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) lifecycle = wrapped_agent._approval_state_store.lifecycle - approval_thread_id = _unscoped_approval_thread_id("thread-legacy-reused") + approval_thread_id = _default_approval_thread_id("thread-legacy-reused") lifecycle.register( owner=ApprovalExecutionOwner.LOCAL, thread_id=approval_thread_id, @@ -5798,7 +5815,7 @@ def guarded_tool() -> str: wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) wrapped_agent._approval_state_store.lifecycle.register( owner=ApprovalExecutionOwner.LOCAL, - thread_id=_unscoped_approval_thread_id("thread-legacy-collision"), + thread_id=_default_approval_thread_id("thread-legacy-collision"), interrupt_id="af-call-secret", call_id="provider-real", name="guarded_tool", @@ -5855,7 +5872,7 @@ def guarded_tool() -> str: wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) wrapped_agent._approval_state_store.lifecycle.register( owner=ApprovalExecutionOwner.LOCAL, - thread_id=_unscoped_approval_thread_id("thread-legacy-duplicate"), + thread_id=_default_approval_thread_id("thread-legacy-duplicate"), interrupt_id="af-call-current", call_id="provider-call", name="guarded_tool", @@ -5909,7 +5926,7 @@ def dangerous_tool(value: str) -> str: wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) wrapped_agent._approval_state_store.lifecycle.register( owner=ApprovalExecutionOwner.LOCAL, - thread_id=_unscoped_approval_thread_id("thread-historical-legacy"), + thread_id=_default_approval_thread_id("thread-historical-legacy"), interrupt_id="af-call-current", call_id="provider-reused", name="dangerous_tool", @@ -6668,7 +6685,7 @@ def record_city(city: str) -> str: ) wrapped_agent = AgentFrameworkAgent(agent=agent, require_confirmation=False) lifecycle = wrapped_agent._approval_state_store.lifecycle - approval_thread_id = _unscoped_approval_thread_id("thread-completed-sibling") + approval_thread_id = _default_approval_thread_id("thread-completed-sibling") completed = lifecycle.register( owner=ApprovalExecutionOwner.LOCAL, thread_id=approval_thread_id, @@ -6925,7 +6942,7 @@ async def test_endpoint_agent_approval_resume_distinguishes_hidden_siblings_with assert {event["toolCallId"] for event in tool_results} == {"provider-shared"} occurrences = wrapped_agent._approval_state_store.lifecycle.occurrences_for_thread( - thread_id=_unscoped_approval_thread_id(thread_id) + thread_id=_default_approval_thread_id(thread_id) ) assert {occurrence.identity.interrupt_id for occurrence in occurrences} == { "guarded-occurrence", @@ -7943,7 +7960,7 @@ async def test_endpoint_agent_approval_cancel_clears_queued_state_when_visible_e approval_id = interrupts[0]["id"] assert approval_id.startswith("af-call-") assert interrupts[0]["toolCallId"] == "call_first" - approval_thread_id = _unscoped_approval_thread_id("thread-queued-cancel-evicted") + approval_thread_id = _default_approval_thread_id("thread-queued-cancel-evicted") stored_state = wrapped_agent._approval_state_store.get_tool_approval_state(approval_thread_id) assert stored_state is not None assert "call_second" in json.dumps(stored_state) @@ -10168,7 +10185,7 @@ async def test_endpoint_builder_checkpoint_storage_rejects_different_thread_afte assert pending_checkpoints checkpoint = max(pending_checkpoints, key=lambda item: item.timestamp) assert checkpoint.metadata[_CHECKPOINT_REQUEST_OWNER_KEY] == { - "snapshot_scope": None, + "snapshot_scope": _DEFAULT_AUTHORIZATION_SCOPE, "thread_id": "victim-thread", } @@ -10228,7 +10245,7 @@ async def test_endpoint_builder_checkpoint_storage_ignores_stale_runtime_owner() assert pending_checkpoints checkpoint = max(pending_checkpoints, key=lambda item: item.timestamp) assert checkpoint.metadata[_CHECKPOINT_REQUEST_OWNER_KEY] == { - "snapshot_scope": None, + "snapshot_scope": _DEFAULT_AUTHORIZATION_SCOPE, "thread_id": "current-thread", } @@ -10253,7 +10270,7 @@ async def test_endpoint_workflow_checkpoint_resume_with_reported_thread_id(): pending_checkpoints = [checkpoint for checkpoint in checkpoints if checkpoint.pending_request_info_events] checkpoint = max(pending_checkpoints, key=lambda item: item.timestamp) assert checkpoint.metadata[_CHECKPOINT_REQUEST_OWNER_KEY] == { - "snapshot_scope": None, + "snapshot_scope": _DEFAULT_AUTHORIZATION_SCOPE, "thread_id": None, "generated_thread_id": thread_id, } @@ -10971,6 +10988,33 @@ async def resolve_authorization_scope(principal: Annotated[str, Depends(authenti assert authenticated == ["principal-a"] +async def test_endpoint_authorization_scope_resolver_awaits_sync_dependency_result(build_chat_client): + """A synchronous FastAPI dependency may return an awaitable scope.""" + app = FastAPI() + agent = Agent(name="test", instructions="Test agent", client=build_chat_client()) + completed: list[str] = [] + + async def resolve_later() -> str: + completed.append("tenant-a") + return "tenant-a" + + def resolve_authorization_scope() -> Awaitable[str]: + return resolve_later() + + add_agent_framework_fastapi_endpoint( + app, + agent, + path="/awaitable-scope", + authorization_scope_resolver=resolve_authorization_scope, + ) + client = TestClient(app) + + response = client.post("/awaitable-scope", json={"messages": [{"role": "user", "content": "Hello"}]}) + + assert response.status_code == 200 + assert completed == ["tenant-a"] + + async def test_endpoint_invalid_authorization_scope_fails_before_runner(build_chat_client): """A configured resolver cannot fall back to unscoped operation.""" app = FastAPI() @@ -11021,27 +11065,52 @@ async def test_endpoint_accepts_snapshot_store_with_scope_resolver(build_chat_cl async def test_endpoint_accepts_snapshot_store_in_single_user_mode(build_chat_client): - """A single-user endpoint gets a stable internal scope when snapshot persistence needs one.""" - app = FastAPI() - agent = Agent(name="test", instructions="Test agent", client=build_chat_client()) - store = InMemoryAGUIThreadSnapshotStore() + """Single-user scopes remain stable per registration and distinct across equal route paths.""" + observed_scopes: list[str] = [] + + class RecordingStore(InMemoryAGUIThreadSnapshotStore): + async def save( + self, + *, + scope: str, + thread_id: str, + snapshot: AGUIThreadSnapshot, + ) -> None: + observed_scopes.append(scope) + await super().save(scope=scope, thread_id=thread_id, snapshot=snapshot) + store = RecordingStore() + app_a = FastAPI() + app_b = FastAPI() add_agent_framework_fastapi_endpoint( - app, - agent, - path="/single-user-snapshots", + app_a, + Agent(name="agent-a", instructions="Test agent", client=build_chat_client("A")), + path="/", snapshot_store=store, single_user=True, ) - client = TestClient(app) - - response = client.post( - "/single-user-snapshots", - json={"messages": [{"role": "user", "content": "Hello"}], "thread_id": "thread-1"}, + add_agent_framework_fastapi_endpoint( + app_b, + Agent(name="agent-b", instructions="Test agent", client=build_chat_client("B")), + path="/", + snapshot_store=store, + single_user=True, ) + client_a = TestClient(app_a) + client_b = TestClient(app_b) + payload = { + "messages": [{"role": "user", "content": "Hello"}], + "thread_id": "shared-thread", + } - assert response.status_code == 200 - assert await store.get(scope="ag-ui:single-user:/single-user-snapshots", thread_id="thread-1") is not None + assert client_a.post("/", json=payload).status_code == 200 + scope_a = observed_scopes[-1] + assert client_b.post("/", json=payload).status_code == 200 + scope_b = observed_scopes[-1] + assert client_a.post("/", json=payload).status_code == 200 + + assert scope_a != scope_b + assert observed_scopes[-1] == scope_a @pytest.mark.parametrize("async_resolver", [False, True], ids=["sync", "async"]) @@ -13737,7 +13806,7 @@ async def stream_fn( assert not [event for event in _decode_sse_events(resume_response) if event.get("type") == "RUN_ERROR"] assert local_executions == [] assert not wrapped_agent._approval_state_store.lifecycle.pending_interrupt_ids( - thread_id=_unscoped_approval_thread_id("thread-hosted-approval") + thread_id=_default_approval_thread_id("thread-hosted-approval") ) approval_responses = [ content @@ -13965,7 +14034,7 @@ async def stream_fn( assert resume_response.status_code == 200 assert local_executions == ["Approved draft"] assert not wrapped_agent._approval_state_store.lifecycle.pending_interrupt_ids( - thread_id=_unscoped_approval_thread_id("thread-local-approval") + thread_id=_default_approval_thread_id("thread-local-approval") ) state_snapshots = [ event["snapshot"] for event in _decode_sse_events(resume_response) if event.get("type") == "STATE_SNAPSHOT" From 286a28e8af42abdba757a506585cf7f85c9de671 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Thu, 15 Oct 2026 15:57:35 +0200 Subject: [PATCH 4/4] Python: Document stable single-user AG-UI scopes --- python/packages/ag-ui/README.md | 23 +++++++++++++++++++ .../ag-ui/agent_framework_ag_ui/_endpoint.py | 5 ++-- 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/python/packages/ag-ui/README.md b/python/packages/ag-ui/README.md index 146fff2937c..44b6728bdc3 100644 --- a/python/packages/ag-ui/README.md +++ b/python/packages/ag-ui/README.md @@ -350,6 +350,29 @@ Authentication in the separate `dependencies` list can allow or reject a request 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 diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py b/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py index cb00ddf86c4..a31bd200172 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_endpoint.py @@ -192,8 +192,9 @@ def add_agent_framework_fastapi_endpoint( single_user: Explicitly declare that every caller of this endpoint belongs to one authorization principal. The endpoint receives one opaque scope that remains stable for that registration and separates equal route paths registered by other applications. Use an explicit resolver when scope continuity must survive process - restarts or when storage is shared beyond one registration. Never enable this mode for callers with - distinct authorization. + restarts or when storage is shared beyond one registration. A single-user application can return its own + stable application scope from authorization_scope_resolver. Never enable this mode for callers with distinct + authorization. """ _validate_keepalive_seconds(keepalive_seconds) _validate_detached_run_options(max_detached_runs, detached_run_timeout_seconds)