Adapters

The provider layer: HTTP client, structural protocols, and response types.

Provider

The default OpenAI-compatible HTTP client. Speaks /chat/completions JSON. Uses stdlib urllib by default; switches to httpx.AsyncClient (with connection pooling) when httpx is installed.

executionkit.provider.Provider dataclass

Provider(base_url: str, model: str, api_key: str = '', default_temperature: float = 0.7, default_max_tokens: int = 4096, timeout: float = 120.0)

Universal LLM provider. Posts JSON, parses JSON. No SDK needed.

Works with any OpenAI-compatible endpoint: OpenAI, Azure, Ollama, Together, Groq, GitHub Models, etc.

aclose async

aclose() -> None

Release the underlying HTTP client.

Call this when the provider is no longer needed (or use it as an async context manager instead).

Source code in executionkit/provider.py

async def aclose(self) -> None:
    """Release the underlying HTTP client.

    Call this when the provider is no longer needed (or use it as an
    async context manager instead).
    """
    if self._use_httpx and self._client is not None:
        await self._client.aclose()

complete async

complete(messages: Sequence[dict[str, Any]], *, temperature: float | None = None, max_tokens: int | None = None, tools: Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> LLMResponse

POST to {base_url}/chat/completions and parse the JSON response.

Source code in executionkit/provider.py

async def complete(
    self,
    messages: Sequence[dict[str, Any]],
    *,
    temperature: float | None = None,
    max_tokens: int | None = None,
    tools: Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> LLMResponse:
    """POST to ``{base_url}/chat/completions`` and parse the JSON response."""
    payload: dict[str, Any] = {
        "model": self.model,
        "messages": list(messages),
        "temperature": (
            temperature if temperature is not None else self.default_temperature
        ),
        "max_tokens": (
            max_tokens if max_tokens is not None else self.default_max_tokens
        ),
    }
    if tools:
        payload["tools"] = list(tools)
    payload.update(kwargs)
    data = await self._post("chat/completions", payload)
    return self._parse_response(data)

Protocols

LLMProvider and ToolCallingProvider are @runtime_checkable Protocols. Any object matching the interface satisfies the protocol — no inheritance required.

executionkit.provider.LLMProvider

Bases: Protocol

Structural protocol for any LLM backend.

Any class with a matching complete signature satisfies this protocol via structural subtyping (PEP 544) — no explicit inheritance required.

executionkit.provider.ToolCallingProvider

Bases: LLMProvider, Protocol

Extension of LLMProvider for providers that support tool calling.

The built-in :class:Provider satisfies this protocol via its supports_tools attribute. Pass to :func:react_loop to unlock tool-calling patterns.

Response types

executionkit.provider.LLMResponse dataclass

LLMResponse(content: str, tool_calls: tuple[ToolCall, ...] = tuple(), finish_reason: str = 'stop', usage: MappingProxyType[str, Any] = (lambda: MappingProxyType({}))(), raw: Any = None)

Parsed LLM completion response.

Handles both OpenAI (prompt_tokens / completion_tokens) and Anthropic (input_tokens / output_tokens) usage key formats.

executionkit.provider.ToolCall dataclass

ToolCall(id: str, name: str, arguments: dict[str, Any])

A single tool invocation extracted from an LLM response.

MockProvider

For unit tests. Yields canned responses, tracks all calls, and never makes real HTTP calls.

executionkit._mock.MockProvider dataclass

MockProvider(responses: list[str | LLMResponse] = list(), exception: Exception | None = None)

Test double implementing LLMProvider and ToolCallingProvider.

Accepts a list of responses (strings or LLMResponse objects) and returns them in order, cycling when exhausted. Optionally raises a configured exception to test error paths.

call_count property

call_count: int

Number of calls made so far.

last_call property

last_call: _CallRecord | None

Most recent call record, or None if no calls yet.

complete async

complete(messages: Sequence[dict[str, Any]], *, temperature: float | None = None, max_tokens: int | None = None, tools: Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> LLMResponse

Return the next pre-configured response or raise the configured exception.

Source code in executionkit/_mock.py

async def complete(
    self,
    messages: Sequence[dict[str, Any]],
    *,
    temperature: float | None = None,
    max_tokens: int | None = None,
    tools: Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> LLMResponse:
    """Return the next pre-configured response or raise the configured exception."""
    self.calls.append(
        _CallRecord(
            messages=list(messages),
            temperature=temperature,
            max_tokens=max_tokens,
            tools=list(tools) if tools else None,
            kwargs=kwargs,
        )
    )

    if self.exception is not None:
        raise self.exception

    if not self.responses:
        return LLMResponse(content="")

    raw = self.responses[self._index % len(self.responses)]
    self._index += 1

    if isinstance(raw, LLMResponse):
        return raw
    return LLMResponse(content=raw)

Custom adapter checklist

Implement a custom provider in three steps:

1. Define a class with an async complete method matching LLMProvider:

from executionkit.provider import LLMResponse

class MyProvider:
    async def complete(
        self,
        messages,
        *,
        temperature=None,
        max_tokens=None,
        tools=None,
        **kwargs,
    ) -> LLMResponse:
        ...

1. Return LLMResponse(content=..., usage={...}). usage should be a dict with at least input_tokens and output_tokens so cost tracking works. Empty dict is acceptable (cost will be 0).

2. For tool calling, set supports_tools = True and populate LLMResponse.tool_calls from the upstream response. react_loop will refuse providers without supports_tools=True.

The structural-protocol design means no registration step — pass your provider directly to any pattern.

Notes on the default Provider

• API key masking. Provider.__repr__ always shows api_key='***' regardless of the actual key length or prefix. Keys are never written to repr output, log lines, or exception messages.
• Credential redaction in errors. HTTP error messages are scanned for credential-shaped substrings (matching sk-..., bearer ..., token=..., etc.) and redacted to [REDACTED] before being raised.
• Connection lifecycle. Provider supports async with and await provider.aclose(). With the httpx backend, this closes the underlying AsyncClient cleanly.
• Retries are at the call layer, not the HTTP layer. Use RetryConfig on the pattern call (e.g. consensus(..., retry=RetryConfig(...))).