mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
731f4fbae6
Add ProviderTransport ABC (4 abstract methods: convert_messages, convert_tools, build_kwargs, normalize_response) plus optional hooks (validate_response, extract_cache_stats, map_finish_reason). Add transport registry with lazy discovery — get_transport() auto-imports transport modules on first call. Add AnthropicTransport — delegates to existing anthropic_adapter.py functions, wired to ALL Anthropic code paths in run_agent.py: - Main normalize loop (L10775) - Main build_kwargs (L6673) - Response validation (L9366) - Finish reason mapping (L9534) - Cache stats extraction (L9827) - Truncation normalize (L9565) - Memory flush build_kwargs + normalize (L7363, L7395) - Iteration-limit summary + retry (L8465, L8498) Zero direct adapter imports remain for transport methods. Client lifecycle, streaming, auth, and credential management stay on AIAgent. 20 new tests (ABC contract, registry, AnthropicTransport methods). 359 anthropic-related tests pass (0 failures). PR 3 of the provider transport refactor.
39 lines
1.2 KiB
Python
39 lines
1.2 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
|