refactor(ai): route Agent.prompt()/stream() through a Transport strategy

The fake, record, inline-fake, and real branches each duplicated the
_log_call() + _apply_schema() tail. Resolve the source once via a Transport
(live / stand-in / inline fake) selected in _transport(), so each public
method has a single call path and logs/applies its schema in one place.

Pure internal refactor: the public prompt/stream/fake/record API and the
fake/record mechanism are unchanged.

Author: tmgbedu

fastapi_startkit/src/fastapi_startkit/ai/agent.py

@@ -6,6 +6,7 @@
 from .document import Document
 from .response import AgentResponse, AgentSnapshot
 from .testing import AgentBinding
+from .transport import InlineFakeTransport, LiveTransport, StandInTransport, Transport
 
 if TYPE_CHECKING:
     from langchain_core.tools import BaseTool
@@ -49,31 +50,12 @@ async def prompt(
         attachments: list[Document] | None = None,
         provider_options: dict | None = None,
     ) -> AgentResponse:
-        stand_in = self._faked()
-        if stand_in is not None:
-            response = await stand_in.prompt(message, attachments=attachments)
-            self._log_call("prompt", message)
-            return self._apply_schema(response)
-
-        _run_kwargs = dict(
+        response = await self._transport(message).prompt(
+            message,
             model=model,
             attachments=attachments,
             provider_options=provider_options,
         )
-
-        match = self._match_fake(message)
-        if match is not None:
-            if isinstance(match, AgentSnapshot):
-                response = await match.resolve(self, message, **_run_kwargs)
-            else:
-                response = match
-            self._log_call("prompt", message)
-            return self._apply_schema(response)
-
-        messages = self._build_messages(message, attachments)
-        chat_model = self._build_model(model, provider_options)
-
-        response = await self._run_pipeline(chat_model, messages)
         self._log_call("prompt", message)
         return self._apply_schema(response)
 
@@ -85,26 +67,11 @@ async def stream(
         provider_options: dict | None = None,
     ) -> AsyncIterator[str]:
         self._log_call("stream", message)
-
-        swapped = self._faked()
-        if swapped is not None:
-            if hasattr(swapped, "stream"):
-                async for chunk in swapped.stream(message):
-                    yield chunk
-            else:
-                response = await swapped.prompt(message)
-                yield response.content
-            return
-
-        fake = self._match_fake(message)
-        if fake is not None:
-            if isinstance(fake, AgentSnapshot):
-                response = await fake.resolve(self, message)
-            else:
-                response = fake
-            yield response.content
-            return
-        async for chunk in self._stream(message, model=model, provider_options=provider_options):
+        async for chunk in self._transport(message).stream(
+            message,
+            model=model,
+            provider_options=provider_options,
+        ):
             yield chunk
 
     @classmethod
@@ -135,6 +102,17 @@ def _faked(self) -> Any:
         binding = type(self)._binding()
         return binding if binding is not self else None
 
+    def _transport(self, message: str) -> Transport:
+        stand_in = self._faked()
+        if stand_in is not None:
+            return StandInTransport(self, stand_in)
+
+        match = self._match_fake(message)
+        if match is not None:
+            return InlineFakeTransport(self, match)
+
+        return LiveTransport(self)
+
     def assert_prompted(self, times: int | None = None) -> None:
         calls = [c for c in self._call_log if c["method"] in ("prompt", "stream")]
         if times is not None:

fastapi_startkit/src/fastapi_startkit/ai/transport.py

@@ -0,0 +1,142 @@
+"""Transport strategies for Agent.prompt()/stream().
+
+A single call resolves exactly one transport — the real model pipeline, a
+stand-in bound via ``Agent.fake()``/``Agent.record()``, or an inline fake matched
+from ``Agent._fakes``. Concentrating the branch here lets the public methods keep
+one call path and log / apply their schema in one place instead of once per branch.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, AsyncIterator
+
+from .response import AgentResponse, AgentSnapshot
+
+if TYPE_CHECKING:
+    from .agent import Agent
+    from .document import Document
+
+
+class Transport:
+    def __init__(self, agent: Agent) -> None:
+        self._agent = agent
+
+    async def prompt(
+        self,
+        message: str,
+        *,
+        model: str | None,
+        attachments: list[Document] | None,
+        provider_options: dict | None,
+    ) -> AgentResponse:
+        raise NotImplementedError
+
+    def stream(
+        self,
+        message: str,
+        *,
+        model: str | None,
+        provider_options: dict | None,
+    ) -> AsyncIterator[str]:
+        raise NotImplementedError
+
+
+class LiveTransport(Transport):
+    """Runs the real model pipeline."""
+
+    async def prompt(
+        self,
+        message: str,
+        *,
+        model: str | None,
+        attachments: list[Document] | None,
+        provider_options: dict | None,
+    ) -> AgentResponse:
+        agent = self._agent
+        messages = agent._build_messages(message, attachments)
+        chat_model = agent._build_model(model, provider_options)
+        return await agent._run_pipeline(chat_model, messages)
+
+    async def stream(
+        self,
+        message: str,
+        *,
+        model: str | None,
+        provider_options: dict | None,
+    ) -> AsyncIterator[str]:
+        async for chunk in self._agent._stream(message, model=model, provider_options=provider_options):
+            yield chunk
+
+
+class StandInTransport(Transport):
+    """Delegates to a stand-in bound via Agent.fake()/record()."""
+
+    def __init__(self, agent: Agent, stand_in: Any) -> None:
+        super().__init__(agent)
+        self._stand_in = stand_in
+
+    async def prompt(
+        self,
+        message: str,
+        *,
+        model: str | None,
+        attachments: list[Document] | None,
+        provider_options: dict | None,
+    ) -> AgentResponse:
+        return await self._stand_in.prompt(message, attachments=attachments)
+
+    async def stream(
+        self,
+        message: str,
+        *,
+        model: str | None,
+        provider_options: dict | None,
+    ) -> AsyncIterator[str]:
+        stand_in = self._stand_in
+        if hasattr(stand_in, "stream"):
+            async for chunk in stand_in.stream(message):
+                yield chunk
+        else:
+            response = await stand_in.prompt(message)
+            yield response.content
+
+
+class InlineFakeTransport(Transport):
+    """Serves a canned response matched from Agent._fakes."""
+
+    def __init__(self, agent: Agent, match: AgentResponse | AgentSnapshot) -> None:
+        super().__init__(agent)
+        self._match = match
+
+    async def prompt(
+        self,
+        message: str,
+        *,
+        model: str | None,
+        attachments: list[Document] | None,
+        provider_options: dict | None,
+    ) -> AgentResponse:
+        match = self._match
+        if isinstance(match, AgentSnapshot):
+            return await match.resolve(
+                self._agent,
+                message,
+                model=model,
+                attachments=attachments,
+                provider_options=provider_options,
+            )
+        return match
+
+    async def stream(
+        self,
+        message: str,
+        *,
+        model: str | None,
+        provider_options: dict | None,
+    ) -> AsyncIterator[str]:
+        match = self._match
+        if isinstance(match, AgentSnapshot):
+            response = await match.resolve(self._agent, message)
+        else:
+            response = match
+        yield response.content