@@ -3520,6 +3520,24 @@ async def __anext__(self) -> UpdateT:
                     update = await update
             return await self._record_update(update)
 
+    async def close(self) -> None:
+        """Close the active iterator and run cleanup hooks.
+
+        This method is idempotent and also closes nested ``ResponseStream`` wrappers.
+        """
+        try:
+            iterator: AsyncIterator[UpdateT] | None = self._iterator
+            if iterator is not None:
+                if isinstance(iterator, ResponseStream):
+                    await cast(ResponseStream[UpdateT, Any], iterator).close()
+                else:
+                    close = getattr(iterator, "aclose", None)
+                    if close is not None:
+                        await close()
+        finally:
+            self._consumed = True
+            await self._run_cleanup_hooks()
+
     async def _resolve_stream_with_pull_contexts(self) -> AsyncIterable[UpdateT]:
         """Resolve the underlying stream while activating any registered pull context managers.
 
 
@@ -49,6 +49,7 @@
 from copy import deepcopy
 from typing import Any, Generic, Literal, TypeVar, overload
 
+from .._agents import BaseAgent
 from .._feature_stage import ExperimentalFeature, experimental
 from .._serialization import make_json_safe
 from .._types import AgentResponse, AgentResponseUpdate, ResponseStream
@@ -1407,7 +1408,7 @@ def _decorator(fn: Callable[..., Awaitable[Any]]) -> FunctionalWorkflowDefinitio
 
 
 @experimental(feature_id=ExperimentalFeature.FUNCTIONAL_WORKFLOWS)
-class FunctionalWorkflowAgent:
+class FunctionalWorkflowAgent(BaseAgent):
     """Agent adapter for a :class:`FunctionalWorkflow`.
 
     Provides a ``run()`` method with the same overloaded signature as
@@ -1452,10 +1453,13 @@ def __init__(
         # but not otherwise consumed.
         del kwargs
         self._workflow = workflow
-        self.name = name or workflow.name
-        self.id = f"FunctionalWorkflowAgent_{self.name}"
-        self.description: str | None = description if description is not None else workflow.description
-        self.context_providers: Sequence[Any] | None = context_providers
+        resolved_name = name or workflow.name
+        super().__init__(
+            id=f"FunctionalWorkflowAgent_{resolved_name}",
+            name=resolved_name,
+            description=description if description is not None else workflow.description,
+            context_providers=context_providers,
+        )
         self._pending_requests: dict[str, WorkflowEvent[Any]] = {}
 
     @property
 
@@ -4192,6 +4192,48 @@ async def async_hook(update: ChatResponseUpdate) -> ChatResponseUpdate:
 class TestResponseStreamCleanupHooks:
     """Tests for cleanup hooks (after stream consumption, before finalizer)."""
 
+    async def test_close_closes_iterator_and_runs_cleanup_once(self) -> None:
+        """Closing a partially consumed stream releases its iterator and cleanup hooks."""
+        events: list[str] = []
+
+        async def updates() -> AsyncIterable[ChatResponseUpdate]:
+            try:
+                yield ChatResponseUpdate(contents=[Content.from_text("first")], role="assistant")
+                yield ChatResponseUpdate(contents=[Content.from_text("second")], role="assistant")
+            finally:
+                events.append("iterator")
+
+        stream: ResponseStream[ChatResponseUpdate, Sequence[ChatResponseUpdate]] = ResponseStream(
+            updates(), cleanup_hooks=[lambda: events.append("cleanup")]
+        )
+        await anext(stream)
+
+        await stream.close()
+        await stream.close()
+
+        assert events == ["iterator", "cleanup"]
+
+    async def test_close_closes_wrapped_stream(self) -> None:
+        """Closing a wrapper releases the concrete inner stream."""
+        events: list[str] = []
+
+        async def updates() -> AsyncIterable[ChatResponseUpdate]:
+            try:
+                yield ChatResponseUpdate(contents=[Content.from_text("first")], role="assistant")
+                yield ChatResponseUpdate(contents=[Content.from_text("second")], role="assistant")
+            finally:
+                events.append("iterator")
+
+        inner: ResponseStream[ChatResponseUpdate, Sequence[ChatResponseUpdate]] = ResponseStream(
+            updates(), cleanup_hooks=[lambda: events.append("inner")]
+        )
+        outer = inner.map(lambda update: update, _combine_updates).with_cleanup_hook(lambda: events.append("outer"))
+        await anext(outer)
+
+        await outer.close()
+
+        assert events == ["iterator", "inner", "outer"]
+
     async def test_cleanup_hook_called_after_iteration(self) -> None:
         """Cleanup hook is called after iteration completes."""
         cleanup_called = {"value": False}
 
@@ -26,6 +26,7 @@
     InMemoryCheckpointStorage,
     RunContext,
     StepWrapper,
+    SupportsAgentRun,
     WorkflowEvent,
     WorkflowEventSource,
     WorkflowRunResult,
@@ -1051,6 +1052,13 @@ async def wf(x: int) -> int:
         assert agent.id == "FunctionalWorkflowAgent_my_agent"
         assert agent.description == "A test workflow"
 
+    async def test_as_agent_implements_supports_agent_run(self):
+        @built_workflow
+        async def wf(x: int) -> int:
+            return x
+
+        assert isinstance(wf.as_agent(), SupportsAgentRun)
+
 
 # ---------------------------------------------------------------------------
 # Concurrent execution guard
 
@@ -2,6 +2,33 @@
 
 This package provides the integration of Agent Framework agents and workflows with the Foundry Agent Server, which can be hosted on Foundry infrastructure.
 
+## Agent instances and factories
+
+`ResponsesHostServer` and `InvocationsHostServer` accept an agent instance or a zero-argument callable through the
+existing `agent` parameter. The callable may be synchronous or asynchronous, must return an object implementing
+`SupportsAgentRun`, and is invoked once for each request:
+
+```python
+server = ResponsesHostServer(agent=create_agent)
+```
+
+Passing an instance reuses that object for the lifetime of the host. Pass a callable when the agent keeps mutable state
+outside `AgentSession`. In particular, a `WorkflowAgent` wraps a stateful workflow, so its callable should build a new
+workflow, executors, and wrapped agents. Keep the workflow name and executor IDs stable so later requests can find and
+restore its checkpoints:
+
+```python
+def create_agent():
+    return build_workflow().as_agent()
+
+
+server = ResponsesHostServer(agent=create_agent)
+```
+
+The Responses host continues regular agents through its existing session store and workflow agents through its
+existing checkpoint store. A callable does not make arbitrary instance fields persistent; state needed by later
+requests must remain in the supported stores.
+
 ## Conversation history
 
 `ResponsesHostServer` uses AgentServer response history as the model's conversation history by default:
@@ -21,8 +48,9 @@ AgentServer history requires a framework `RawAgent` whose client declares the bo
 the agent's runtime options then let hosting enforce downstream storage behavior. Custom `SupportsAgentRun`
 implementations must use `history_source="agent"` because that protocol does not accept runtime chat options.
 
-`ResponsesHostServer` owns the supplied agent instance and may add hosting-specific context providers. Do not reuse that
-agent with another host or invoke it directly after constructing the server.
+`ResponsesHostServer` owns a supplied agent instance and may add hosting-specific context providers. Do not reuse that
+instance with another host or invoke it directly after constructing the server. An agent returned by a callable belongs
+to that request.
 
 To preserve the agent's regular history and service-storage behavior, select the agent as the history source:
 
 
@@ -0,0 +1,39 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+import inspect
+from collections.abc import Awaitable, Callable
+from typing import TypeAlias, TypeGuard, cast
+
+from agent_framework import SupportsAgentRun
+
+AgentSource: TypeAlias = SupportsAgentRun | Callable[[], SupportsAgentRun | Awaitable[SupportsAgentRun]]
+
+
+def is_agent(value: object) -> TypeGuard[SupportsAgentRun]:
+    return not inspect.isclass(value) and isinstance(value, SupportsAgentRun)
+
+
+def validate_agent_source(source: object) -> None:
+    if is_agent(source):
+        return
+    if not callable(source):
+        raise TypeError("agent must be an agent instance or a zero-argument callable that creates one.")
+    try:
+        inspect.signature(source).bind()
+    except (TypeError, ValueError) as exc:
+        raise TypeError("agent callable must accept no arguments.") from exc
+
+
+async def resolve_agent(source: AgentSource) -> SupportsAgentRun:
+    """Resolve an agent instance or request-scoped agent factory."""
+    if is_agent(source):
+        return source
+
+    factory = cast(Callable[[], SupportsAgentRun | Awaitable[SupportsAgentRun]], source)
+    result = factory()
+    agent = await result if inspect.isawaitable(result) else result
+    if not is_agent(agent):
+        raise TypeError("The agent factory must return an object implementing SupportsAgentRun.")
+    return agent
@@ -1,15 +1,18 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import json
+from collections.abc import Awaitable, Callable
+from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
 
-from agent_framework import AgentSession, SupportsAgentRun
+from agent_framework import AgentSession, ResponseStream, SupportsAgentRun
 from agent_framework._telemetry import mark_feature_used
 from azure.ai.agentserver.core import get_request_context
 from azure.ai.agentserver.invocations import InvocationAgentServerHost
 from starlette.requests import Request
 from starlette.responses import Response, StreamingResponse
 from typing_extensions import Any, AsyncGenerator
 
+from ._agent_source import is_agent, resolve_agent, validate_agent_source
 from ._feature_usage import FeatureIndex
 
 
@@ -18,25 +21,28 @@ class InvocationsHostServer(InvocationAgentServerHost):
 
     def __init__(
         self,
-        agent: SupportsAgentRun,
+        agent: SupportsAgentRun | Callable[[], SupportsAgentRun | Awaitable[SupportsAgentRun]],
         *,
         openapi_spec: dict[str, Any] | None = None,
         **kwargs: Any,
     ) -> None:
         """Initialize an InvocationsHostServer.
 
         Args:
-            agent: The agent to handle responses for.
+            agent: The agent to handle responses for, or a zero-argument sync or async callable that creates one for
+                each request. Use a callable for agents that keep mutable state outside `AgentSession`.
             openapi_spec: The OpenAPI specification for the server.
             **kwargs: Additional keyword arguments.
 
         This host will expect the request to be a JSON body with a "message" field.
         The response from the host will be a JSON object with a "response" field containing
         the agent's response and a "session_id" field containing the session ID.
         """
+        validate_agent_source(agent)
         super().__init__(openapi_spec=openapi_spec, **kwargs)
 
         self._agent = agent
+        self._owns_request_agent = not is_agent(agent)
         self._sessions: dict[str | tuple[str, str], AgentSession] = {}
         self.invoke_handler(self._handle_invoke)
         mark_feature_used(FeatureIndex.FOUNDRY_HOSTING)
@@ -72,6 +78,14 @@ def _partition_key(self) -> str | tuple[str, str]:
 
         return context.session_id
 
+    @asynccontextmanager
+    async def _request_agent(self) -> AsyncGenerator[SupportsAgentRun]:
+        agent = await resolve_agent(self._agent)
+        async with AsyncExitStack() as resources:
+            if self._owns_request_agent and isinstance(agent, AbstractAsyncContextManager):
+                await resources.enter_async_context(agent)
+            yield agent
+
     async def _handle_invoke(self, request: Request) -> Response:
         """Invoke the agent with the given request."""
         try:
@@ -100,15 +114,26 @@ async def _handle_invoke(self, request: Request) -> Response:
         if stream:
 
             async def stream_response() -> AsyncGenerator[str]:
-                async for update in self._agent.run(user_message, session=session, stream=True):
-                    if update.text:
-                        yield update.text
+                async with self._request_agent() as agent:
+                    stream = agent.run(user_message, session=session, stream=True)
+                    try:
+                        async for update in stream:
+                            if update.text:
+                                yield update.text
+                    finally:
+                        if isinstance(stream, ResponseStream):
+                            await stream.close()
+                        else:
+                            close = getattr(stream, "aclose", None)
+                            if close is not None:
+                                await close()
 
             return StreamingResponse(
                 stream_response(),
                 media_type="text/event-stream",
                 headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
             )
 
-        response = await self._agent.run([user_message], session=session)
+        async with self._request_agent() as agent:
+            response = await agent.run([user_message], session=session)
         return Response(content=response.text)