c832ebd67c
Add ResponsesApiTransport wrapping codex_responses_adapter.py behind the ProviderTransport ABC. Auto-registered via _discover_transports(). Wire ALL Codex transport methods to production paths in run_agent.py: - build_kwargs: main _build_api_kwargs codex branch (50 lines extracted) - normalize_response: main loop + flush + summary + retry (4 sites) - convert_tools: memory flush tool override - convert_messages: called internally via build_kwargs - validate_response: response validation gate - preflight_kwargs: request sanitization (2 sites) Remove 7 dead legacy wrappers from AIAgent (_responses_tools, _chat_messages_to_responses_input, _normalize_codex_response, _preflight_codex_api_kwargs, _preflight_codex_input_items, _extract_responses_message_text, _extract_responses_reasoning_text). Keep 3 ID manipulation methods still used by _build_assistant_message. Update 18 test call sites across 3 test files to call adapter functions directly instead of through deleted AIAgent wrappers. 24 new tests. 343 codex/responses/transport tests pass (0 failures). PR 4 of the provider transport refactor.
44 lines
1.3 KiB
Python
44 lines
1.3 KiB
Python
"""Transport layer types and registry for provider response normalization.
|
|
|
|
Usage:
|
|
from agent.transports import get_transport
|
|
transport = get_transport("anthropic_messages")
|
|
result = transport.normalize_response(raw_response)
|
|
"""
|
|
|
|
from agent.transports.types import NormalizedResponse, ToolCall, Usage, build_tool_call, map_finish_reason # noqa: F401
|
|
|
|
_REGISTRY: dict = {}
|
|
|
|
|
|
def register_transport(api_mode: str, transport_cls: type) -> None:
|
|
"""Register a transport class for an api_mode string."""
|
|
_REGISTRY[api_mode] = transport_cls
|
|
|
|
|
|
def get_transport(api_mode: str):
|
|
"""Get a transport instance for the given api_mode.
|
|
|
|
Returns None if no transport is registered for this api_mode.
|
|
This allows gradual migration — call sites can check for None
|
|
and fall back to the legacy code path.
|
|
"""
|
|
if not _REGISTRY:
|
|
_discover_transports()
|
|
cls = _REGISTRY.get(api_mode)
|
|
if cls is None:
|
|
return None
|
|
return cls()
|
|
|
|
|
|
def _discover_transports() -> None:
|
|
"""Import all transport modules to trigger auto-registration."""
|
|
try:
|
|
import agent.transports.anthropic # noqa: F401
|
|
except ImportError:
|
|
pass
|
|
try:
|
|
import agent.transports.codex # noqa: F401
|
|
except ImportError:
|
|
pass
|