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)

    with llm_span(self.model) as span:
        data = await self._post("chat/completions", payload)
        response = self._parse_response(data)
        record_llm_span_attributes(
            span,
            self.model,
            response.input_tokens,
            response.output_tokens,
        )
        return response

stream

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

Stream a completion as live text deltas over OpenAI-compatible SSE.

Sends stream: true plus stream_options.include_usage so the server emits a final usage frame. Yields choices[0].delta.content strings as they arrive. When usage_sink is supplied, the final :class:LLMResponse (carrying token usage) is appended to it once the stream drains, so budget-aware callers can record cost afterwards.

This is a regular method that returns an async iterator — call it without await and consume it with async for.

Source code in executionkit/provider.py

def stream(
    self,
    messages: Sequence[dict[str, Any]],
    *,
    temperature: float | None = None,
    max_tokens: int | None = None,
    tools: Sequence[dict[str, Any]] | None = None,
    usage_sink: list[LLMResponse] | None = None,
    **kwargs: Any,
) -> AsyncIterator[str]:
    """Stream a completion as live text deltas over OpenAI-compatible SSE.

    Sends ``stream: true`` plus ``stream_options.include_usage`` so the
    server emits a final usage frame.  Yields ``choices[0].delta.content``
    strings as they arrive.  When *usage_sink* is supplied, the final
    :class:`LLMResponse` (carrying token usage) is appended to it once the
    stream drains, so budget-aware callers can record cost afterwards.

    This is a regular method that *returns* an async iterator — call it
    without ``await`` and consume it with ``async for``.
    """
    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
        ),
        "stream": True,
        "stream_options": {"include_usage": True},
    }
    if tools:
        payload["tools"] = list(tools)
    payload.update(kwargs)
    return self._stream(payload, usage_sink)

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.

The library redacts content before emitting it in any trace it owns. raw is the verbatim, unredacted provider payload and is therefore caller-owned: the library never emits it, and a caller that logs or traces raw is responsible for redacting any credentials it may contain.

executionkit.provider.ToolCall dataclass

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

A single tool invocation extracted from an LLM response.

__post_init__

__post_init__() -> None

Wrap arguments in a read-only proxy to enforce immutability.

Source code in executionkit/provider.py

def __post_init__(self) -> None:
    """Wrap ``arguments`` in a read-only proxy to enforce immutability."""
    if not isinstance(self.arguments, MappingProxyType):
        object.__setattr__(
            self, "arguments", MappingProxyType(dict(self.arguments))
        )

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)

stream

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

Stream the next canned response one character at a time.

Mirrors :meth:complete accounting (records the call, honours exception, cycles the response index) but yields the response incrementally. When usage_sink is provided, a final :class:LLMResponse carrying synthesized token usage is appended so streaming callers can record non-zero output tokens after draining.

Source code in executionkit/_mock.py

def stream(
    self,
    messages: Sequence[dict[str, Any]],
    *,
    temperature: float | None = None,
    max_tokens: int | None = None,
    tools: Sequence[dict[str, Any]] | None = None,
    usage_sink: list[LLMResponse] | None = None,
    **kwargs: Any,
) -> AsyncIterator[str]:
    """Stream the next canned response one character at a time.

    Mirrors :meth:`complete` accounting (records the call, honours
    ``exception``, cycles the response index) but yields the response
    incrementally.  When *usage_sink* is provided, a final
    :class:`LLMResponse` carrying synthesized token usage is appended so
    streaming callers can record non-zero output tokens after draining.
    """
    self.calls.append(
        _CallRecord(
            messages=list(messages),
            temperature=temperature,
            max_tokens=max_tokens,
            tools=list(tools) if tools else None,
            kwargs={**kwargs, "stream": True},
        )
    )
    return self._stream(usage_sink)

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(...))).