feat: capture the Responses API, and any framework through a capturing HTTP transport

README.md

@@ -87,6 +87,32 @@ with agentcompile.conversation(ticket.id, customer=ticket.customer_id):
 Streamed answers are captured too: your agent reads the stream as always, and the whole answer
 is captured once it ends.
 
+## The Responses API and any framework
+
+A wrapped OpenAI client captures `client.responses.create(...)` as well (streamed or not). Those
+calls always go to your model: compiled answers come in Chat Completions' and Anthropic's shapes.
+
+When a framework builds its own client (LangChain, LiteLLM, CrewAI, Pydantic AI, the OpenAI
+Agents SDK), give it a capturing HTTP client instead of wrapping anything:
+
+```python
+http = agentcompile.http_client(key="ack_...")         # async: agentcompile.async_http_client()
+
+ChatOpenAI(model="gpt-5", http_client=http)            # LangChain
+litellm.client_session = http                          # LiteLLM, CrewAI
+OpenAIProvider(http_client=agentcompile.async_http_client())   # Pydantic AI
+set_default_openai_client(AsyncOpenAI(http_client=agentcompile.async_http_client()))  # Agents SDK
+
+with agentcompile.conversation(ticket.id):
+    run_agent()
+```
+
+It captures the POSTs to `/chat/completions`, `/responses` and `/messages` that pass through
+it, scrubbed the same way, and leaves everything else alone. It only captures: to have known
+jobs answered compiled, use `wrap`. An SDK that takes only `httpx2` clients (newer `anthropic`
+releases) takes `agentcompile.http_client(lib="httpx2")`.
+`agentcompile.transport()` gives the bare transport, to wrap one of your own.
+
 ## Privacy
 
 With `capture=True`, personal data is scrubbed on your machine before anything is sent:

src/agentcompile/__init__.py

@@ -17,15 +17,40 @@
 
 from ._capture import capturer_for, flush_all
 from ._conversation import conversation
-from ._core import LIVE, SHADOW, Proxy, Router, async_create, sync_create
+from ._core import (
+    LIVE,
+    SHADOW,
+    Proxy,
+    Router,
+    async_create,
+    async_forward,
+    sync_create,
+    sync_forward,
+)
 from ._decide import AsyncDecider, Decider, Settings
 from ._outcome import OUTCOMES, outcome
 from ._outcome import flush as _flush_outcomes
 from ._outcome import remember as _remember
 from ._scrub import load_key
 from ._trail import OnEvent, Trail
-
-__all__ = ["LIVE", "OUTCOMES", "SHADOW", "Settings", "conversation", "flush", "outcome", "wrap"]
+from ._transport import AsyncCaptureTransport, CaptureTransport, classes, lib_of
+
+__all__ = [
+    "LIVE",
+    "OUTCOMES",
+    "SHADOW",
+    "AsyncCaptureTransport",
+    "CaptureTransport",
+    "Settings",
+    "async_http_client",
+    "async_transport",
+    "conversation",
+    "flush",
+    "http_client",
+    "outcome",
+    "transport",
+    "wrap",
+]
 try:
     from ._version import __version__
 except ImportError:  # running from a source checkout that was never built
@@ -60,13 +85,7 @@ def wrap(
     this machine before anything is sent (default True; the key is AGENTCOMPILE_SCRUB_KEY or
     one created once in ~/.agentcompile/).
     """
-    settings = Settings(
-        base_url=base_url or os.environ.get("AGENTCOMPILE_URL", DEFAULT_URL),
-        key=key or os.environ.get("AGENTCOMPILE_KEY"),
-        company=company or os.environ.get("AGENTCOMPILE_COMPANY"),
-        timeout=timeout,
-        mode=mode,
-    )
+    settings = _settings(key, base_url, company, timeout, mode)
     the_trail = Trail(trail, on_event)
     _remember(settings)  # outcome() sends with the client wrapped last
     if capture is None:
@@ -90,10 +109,21 @@ def wrap(
         router = _router("openai", completions.create, settings, the_trail, mode, http_client)
         router.capturer = capturer
         create = _create(router, completions.create, _openai.build)
-        return Proxy(
-            client,
-            {"chat": Proxy(chat, {"completions": Proxy(completions, {"create": create})})},
-        )
+        overrides: dict[str, Any] = {
+            "chat": Proxy(chat, {"completions": Proxy(completions, {"create": create})})
+        }
+        responses = getattr(client, "responses", None)
+        if responses is not None and hasattr(responses, "create"):
+            # The Responses API (the OpenAI Agents SDK's default): captured, always the model's.
+            forward = _router(
+                "openai", responses.create, settings, the_trail, mode, http_client
+            )
+            forward.capturer = capturer
+            make = async_forward if _is_async(responses.create) else sync_forward
+            overrides["responses"] = Proxy(
+                responses, {"create": make(forward, responses.create)}
+            )
+        return Proxy(client, overrides)
 
     messages = getattr(client, "messages", None)
     if messages is not None and hasattr(messages, "create"):
@@ -107,6 +137,84 @@ def wrap(
     raise TypeError("agentcompile.wrap expects an OpenAI- or Anthropic-style client")
 
 
+def transport(
+    wrapped: Any = None,
+    *,
+    lib: str | None = None,
+    key: str | None = None,
+    base_url: str | None = None,
+    company: str | None = None,
+    scrub: bool = True,
+    timeout: float = 2.0,
+) -> Any:
+    """A transport that captures every model call through it, for frameworks whose client you
+    don't hold (LangChain, LiteLLM, CrewAI, Pydantic AI, ...). Capture only: calls are never
+    answered compiled. `wrapped`: the transport to send through (default the library's own).
+    `lib`: "httpx" (default) or "httpx2", the fork newer provider SDKs require; taken from
+    `wrapped` when it is given. The other arguments are wrap()'s; `timeout` bounds each
+    capture send, not your calls."""
+    sync, _ = classes(lib_of(wrapped, lib))
+    return sync(_capturer(key, base_url, company, scrub, timeout), wrapped)
+
+
+def async_transport(
+    wrapped: Any = None,
+    *,
+    lib: str | None = None,
+    key: str | None = None,
+    base_url: str | None = None,
+    company: str | None = None,
+    scrub: bool = True,
+    timeout: float = 2.0,
+) -> Any:
+    """The async version of transport()."""
+    _, async_ = classes(lib_of(wrapped, lib))
+    return async_(_capturer(key, base_url, company, scrub, timeout), wrapped)
+
+
+def http_client(*, lib: str = "httpx", **options: Any) -> Any:
+    """A client whose model calls are captured: pass it as the framework's `http_client`.
+    `lib="httpx2"` for SDKs built on httpx2. Takes transport()'s keyword arguments."""
+    import importlib
+
+    module = importlib.import_module(lib)
+    return module.Client(transport=transport(lib=lib, **options), timeout=_CALL_TIMEOUT)
+
+
+def async_http_client(*, lib: str = "httpx", **options: Any) -> Any:
+    """An async client whose model calls are captured."""
+    import importlib
+
+    module = importlib.import_module(lib)
+    return module.AsyncClient(
+        transport=async_transport(lib=lib, **options), timeout=_CALL_TIMEOUT
+    )
+
+
+# The provider SDKs' own default: a model call can take minutes.
+_CALL_TIMEOUT = 600.0
+
+
+def _settings(
+    key: str | None, base_url: str | None, company: str | None, timeout: float, mode: str
+) -> Settings:
+    return Settings(
+        base_url=base_url or os.environ.get("AGENTCOMPILE_URL", DEFAULT_URL),
+        key=key or os.environ.get("AGENTCOMPILE_KEY"),
+        company=company or os.environ.get("AGENTCOMPILE_COMPANY"),
+        timeout=timeout,
+        mode=mode,
+    )
+
+
+def _capturer(
+    key: str | None, base_url: str | None, company: str | None, scrub: bool, timeout: float
+) -> Any:
+    settings = _settings(key, base_url, company, timeout, LIVE)
+    _remember(settings)
+    return capturer_for(settings, None, load_key() if scrub else None)
+
+
 def flush(timeout: float = 5.0) -> None:
     """Send every captured call and outcome still queued (short scripts and tests; also runs
     at exit)."""

src/agentcompile/_assemble.py

@@ -1,7 +1,7 @@
 """Streamed answers, captured whole: each chunk passes to your agent untouched as it arrives, a
 copy is kept, and when the stream ends the pieces are assembled into the same shape a
-non-streamed answer has (a chat completion, or an Anthropic message), so AgentCompile reads
-streamed and non-streamed calls the same way."""
+non-streamed answer has (a chat completion, a Responses API response, or an Anthropic message),
+so AgentCompile reads streamed and non-streamed calls the same way."""
 
 from __future__ import annotations
 
@@ -16,7 +16,11 @@
 
 def assemble(provider: str, chunks: list[Any]) -> dict[str, Any] | None:
     try:
-        return _openai(chunks) if provider == "openai" else _anthropic(chunks)
+        if provider != "openai":
+            return _anthropic(chunks)
+        if any(str(c.get("type", "")).startswith("response.") for c in chunks):
+            return _responses(chunks)
+        return _openai(chunks)
     except Exception:  # never break the agent over a shape we didn't expect
         return None
 
@@ -60,6 +64,27 @@ def _openai(chunks: list[dict[str, Any]]) -> dict[str, Any]:
     return out
 
 
+#: The events that end a Responses API stream, each carrying the whole response.
+_RESPONSE_DONE = ("response.completed", "response.incomplete", "response.failed")
+
+
+def _responses(events: list[dict[str, Any]]) -> dict[str, Any]:
+    """A Responses API stream: its last event carries the whole response. A stream stopped
+    early has none, so the response is rebuilt from the items it finished."""
+    for event in reversed(events):
+        if event.get("type") in _RESPONSE_DONE and isinstance(event.get("response"), dict):
+            return dict(event["response"])
+    response: dict[str, Any] = {}
+    items: dict[int, dict[str, Any]] = {}
+    for event in events:
+        if event.get("type") == "response.created" and isinstance(event.get("response"), dict):
+            response = dict(event["response"])
+        elif event.get("type") == "response.output_item.done":
+            items[int(event.get("output_index", len(items)))] = dict(event.get("item") or {})
+    response["output"] = [items[i] for i in sorted(items)]
+    return response
+
+
 def _anthropic(events: list[dict[str, Any]]) -> dict[str, Any]:
     message: dict[str, Any] = {}
     blocks: dict[int, dict[str, Any]] = {}

src/agentcompile/_capture.py

@@ -21,7 +21,7 @@
 
 from ._assemble import assemble
 from ._decide import CONVERSATION_HEADER, Settings, _request
-from ._payload import jsonable, payload
+from ._payload import jsonable, request_of
 from ._scrub import scrub_call, scrub_value
 
 MAX_QUEUE = 2000
@@ -88,7 +88,7 @@ def add(
         """Queue one call. Never raises. `stream` without `streamed`: the answer wasn't kept
         (the record says so); `streamed`: the answer was assembled from its stream."""
         try:
-            request, answer = payload(kwargs), None if stream else jsonable(response)
+            request, answer = request_of(kwargs), None if stream else jsonable(response)
             if self.scrub_key is not None:
                 request = scrub_call(request, self.scrub_key)
                 answer = scrub_call(answer, self.scrub_key)

src/agentcompile/_core.py

@@ -250,6 +250,65 @@ async def create(*args: Any, **kwargs: Any) -> Any:
     return create
 
 
+#: Why a Responses API call is never asked about: compiled answers come in Chat Completions' and
+#: Anthropic's shapes, so these calls go to the model, captured like any other.
+RESPONSES_API = "responses api"
+
+
+def _forward_only(router: Router, conversation_id: str | None) -> tuple[str, str | None]:
+    return (
+        ("no-conversation", None) if conversation_id is None else ("unsupported", RESPONSES_API)
+    )
+
+
+def sync_forward(router: Router, original: Callable[..., Any]) -> Callable[..., Any]:
+    """create() for an API we capture but don't answer: straight to the model."""
+
+    def create(*args: Any, **kwargs: Any) -> Any:
+        started = time.perf_counter()
+        conversation_id, real_kwargs = router.prepare(kwargs)
+        stream = bool(real_kwargs.get("stream"))
+        result = original(*args, **real_kwargs)
+        result = router.capture(conversation_id, real_kwargs, result, stream)
+        route, error = _forward_only(router, conversation_id)
+        router.record(
+            route=route,
+            conversation_id=conversation_id,
+            kwargs=real_kwargs,
+            decision=None,
+            decide_ms=None,
+            error=error,
+            started=started,
+            stream=stream,
+        )
+        return result
+
+    return create
+
+
+def async_forward(router: Router, original: Callable[..., Any]) -> Callable[..., Any]:
+    async def create(*args: Any, **kwargs: Any) -> Any:
+        started = time.perf_counter()
+        conversation_id, real_kwargs = router.prepare(kwargs)
+        stream = bool(real_kwargs.get("stream"))
+        result = await original(*args, **real_kwargs)
+        result = router.capture(conversation_id, real_kwargs, result, stream, True)
+        route, error = _forward_only(router, conversation_id)
+        router.record(
+            route=route,
+            conversation_id=conversation_id,
+            kwargs=real_kwargs,
+            decision=None,
+            decide_ms=None,
+            error=error,
+            started=started,
+            stream=stream,
+        )
+        return result
+
+    return create
+
+
 class Proxy:
     """Everything delegates to the real object except the attributes we override."""
 

src/agentcompile/_payload.py

@@ -6,6 +6,15 @@
 
 # What the engine reads; everything else stays between the customer and their provider.
 _FIELDS = ("model", "messages", "system", "tools")
+# A Responses API call (OpenAI's `responses.create`): the same, in that API's own words.
+_RESPONSES_FIELDS = (
+    "model",
+    "input",
+    "instructions",
+    "tools",
+    "previous_response_id",
+    "conversation",
+)
 
 
 def jsonable(value: Any) -> Any:
@@ -24,3 +33,14 @@ def jsonable(value: Any) -> Any:
 
 def payload(kwargs: dict[str, Any]) -> dict[str, Any]:
     return {k: jsonable(kwargs[k]) for k in _FIELDS if k in kwargs and kwargs[k] is not None}
+
+
+def request_of(kwargs: dict[str, Any]) -> dict[str, Any]:
+    """What capture keeps of a call: `payload`, or a Responses API call's own fields."""
+    if "input" in kwargs and "messages" not in kwargs:
+        return {
+            k: jsonable(kwargs[k])
+            for k in _RESPONSES_FIELDS
+            if k in kwargs and kwargs[k] is not None
+        }
+    return payload(kwargs)