Providers and responses¶

Built-in HTTP client¶

Provider appends /chat/completions to base_url, sends OpenAI-format JSON, and parses OpenAI-format responses. It uses urllib in the base install and an httpx.AsyncClient when the httpx extra 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)

HTTP client for OpenAI-compatible chat-completions endpoints.

The client sends and parses JSON without a provider SDK. Endpoint and model compatibility remain the caller's responsibility.

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)

Provider protocols¶

These are structural protocols. A custom provider does not need to inherit from them, but its methods and attributes must have compatible signatures.

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 it to :func:react_loop for tool-calling patterns.

executionkit.provider.StreamingProvider ¶

Bases: LLMProvider, Protocol

Extension of LLMProvider for providers that can stream completions.

A streaming provider yields incremental text deltas via :meth:stream instead of returning a single :class:LLMResponse. stream is a normal (non-async) method that returns an async iterator — call it without await and consume the result with async for. Satisfied structurally (PEP 544); the built-in :class:Provider and the test :class:~executionkit._mock.MockProvider both implement it.

At runtime, a @runtime_checkable protocol checks whether required attributes exist; it does not validate the full type signature. Run a static type checker and integration tests for custom adapters.

Set supports_tools = True only when the adapter can send OpenAI-format tool schemas and parse tool calls. react_loop() rejects a provider when the flag is absent or false.

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

LLMResponse.raw is the unmodified provider payload. ExecutionKit does not emit it in package-owned traces. Caller code must redact it before logging.

LLMResponse.usage accepts these token field names:

prompt_tokens and completion_tokens; or
input_tokens and output_tokens.

Missing usage fields count as zero. Boolean, negative, or implausibly large counts raise ProviderError when the properties are read.

Test provider¶

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)

Import it from the package root:

from executionkit import MockProvider

provider = MockProvider(responses=["draft", "final"])

It records calls and returns scripted responses without network access.

Adapter checklist¶

A custom adapter should:

accept OpenAI-format message dictionaries;
translate pattern options such as temperature, max_tokens, and tools;
return LLMResponse;
translate provider tool calls into ToolCall;
report token usage when available;
map retryable and non-retryable failures to the appropriate ExecutionKit exception; and
test basic calls, tools, streaming, cancellation, timeout, and malformed responses separately.

See Provider setup for a complete custom adapter skeleton.

Providers and responses¶

Built-in HTTP client¶

executionkit.provider.Provider dataclass ¶

aclose async ¶

complete async ¶

stream ¶

Provider protocols¶

executionkit.provider.LLMProvider ¶

executionkit.provider.ToolCallingProvider ¶

executionkit.provider.StreamingProvider ¶

Response types¶

executionkit.provider.LLMResponse dataclass ¶

executionkit.provider.ToolCall dataclass ¶

__post_init__ ¶

Test provider¶

executionkit._mock.MockProvider dataclass ¶

call_count property ¶

last_call property ¶

complete async ¶

stream ¶

Adapter checklist¶

executionkit.provider.Provider `dataclass` ¶

aclose `async` ¶

complete `async` ¶

executionkit.provider.LLMResponse `dataclass` ¶

executionkit.provider.ToolCall `dataclass` ¶

executionkit._mock.MockProvider `dataclass` ¶

call_count `property` ¶

last_call `property` ¶

complete `async` ¶