Skip to content

Protocols

agenticraft_types.protocols

Protocol classes for structural typing across AgentiCraft packages.

Provider

Bases: Protocol

Protocol for LLM provider implementations.

Source code in src/agenticraft_types/protocols.py
@runtime_checkable
class Provider(Protocol):
    """Protocol for LLM provider implementations."""

    @property
    def name(self) -> str: ...

    async def complete(
        self,
        messages: list[dict[str, Any]],
        model: str | None = None,
        **kwargs: Any,
    ) -> CompletionResponse: ...

    def supports_model(self, model: str) -> bool: ...

StreamingProvider

Bases: Protocol

Protocol for providers that support streaming.

Source code in src/agenticraft_types/protocols.py
@runtime_checkable
class StreamingProvider(Protocol):
    """Protocol for providers that support streaming."""

    async def stream(
        self,
        messages: list[dict[str, Any]],
        model: str | None = None,
        **kwargs: Any,
    ) -> Any: ...  # AsyncIterator[StreamChunk]

Router

Bases: Protocol

Protocol for provider routing implementations.

Source code in src/agenticraft_types/protocols.py
@runtime_checkable
class Router(Protocol):
    """Protocol for provider routing implementations."""

    async def route(
        self,
        providers: list[str],
        **kwargs: Any,
    ) -> str | None: ...

    def get_stats(self) -> dict[str, Any]: ...

CircuitBreakerLike

Bases: Protocol

Protocol for circuit breaker implementations.

Source code in src/agenticraft_types/protocols.py
@runtime_checkable
class CircuitBreakerLike(Protocol):
    """Protocol for circuit breaker implementations."""

    @property
    def is_open(self) -> bool: ...

    def record_success(self) -> None: ...

    def record_failure(self) -> None: ...

    def reset(self) -> None: ...

KeyPoolLike

Bases: Protocol

Protocol for API key pool implementations.

Source code in src/agenticraft_types/protocols.py
@runtime_checkable
class KeyPoolLike(Protocol):
    """Protocol for API key pool implementations."""

    async def acquire(self) -> str: ...

    async def record_success(self, key: str, **kwargs: Any) -> None: ...

    async def record_error(self, key: str, **kwargs: Any) -> None: ...