API Reference

Auto-generated from docstrings. For narrative guides, see Guides.

Runtime

byoai.Runtime

Runtime(*, llm: dict[str, Any] | None = None, providers: list[LLMProvider | Callable[..., Any]] | None = None, cache: dict[str, Any] | CacheStore | None = None, vector_store: dict[str, Any] | VectorStore | Callable[..., Any] | None = None, semantic_cache: dict[str, Any] | SemanticCacheStore | None = None, embedder: dict[str, Any] | Embedder | None = None, retry_policy: RetryPolicy | None = None, selection: SelectionName | SelectionFn = 'ordered', system_prompt: str | None = None, telemetry: Any | None = None)

The execution engine: owns middleware, pipelines, events, provider routing, caching and usage accounting. Configure it declaratively (llm={...}, cache={...}) or with pre-built adapter instances; both styles compose.

Source code in src/byoai/runtime.py

def __init__(
    self,
    *,
    llm: dict[str, Any] | None = None,
    providers: list[LLMProvider | Callable[..., Any]] | None = None,
    cache: dict[str, Any] | CacheStore | None = None,
    vector_store: dict[str, Any] | VectorStore | Callable[..., Any] | None = None,
    semantic_cache: dict[str, Any] | SemanticCacheStore | None = None,
    embedder: dict[str, Any] | Embedder | None = None,
    retry_policy: RetryPolicy | None = None,
    selection: SelectionName | SelectionFn = "ordered",
    system_prompt: str | None = None,
    telemetry: Any | None = None,
) -> None:
    self.events = EventBus()
    self.middleware = MiddlewareChain()
    self._pipelines: dict[str, Pipeline] = {}

    self.cache: CacheStore | None = (
        build_cache(cache) if isinstance(cache, dict) else cache
    )
    self.vector_store: VectorStore | None = (
        build_vector_store(vector_store)
        if isinstance(vector_store, dict)
        else FunctionVectorStore(vector_store)  # type: ignore[arg-type]
        if vector_store is not None and not hasattr(vector_store, "search")
        else vector_store
    )

    resolved_providers = list(providers or [])
    if llm is not None:
        resolved_providers = build_router(llm) + resolved_providers
    self.router: ProviderRouter | None = (
        ProviderRouter(
            resolved_providers,
            retry_policy=retry_policy,
            selection=selection,
            event_bus=self.events,
        )
        if resolved_providers
        else None
    )

    self.embedder: Embedder | None = (
        build_embedder(embedder) if isinstance(embedder, dict) else embedder
    )
    self.semantic_cache: SemanticCacheStore | None = (
        build_semantic_cache(semantic_cache)
        if isinstance(semantic_cache, dict)
        else semantic_cache
    )
    if (
        isinstance(semantic_cache, dict)
        # Read the metric the store actually resolved to, not the raw
        # config dict re-defaulted to "cosine" here too: a plugin-
        # provided semantic cache (byoai.semantic_caches entry point, or
        # any future built-in) can default to a non-cosine metric on its
        # own even when the config dict never mentions "metric" at all —
        # re-deriving "cosine" from the dict's absence would silently
        # skip this guard for exactly the case it exists to catch.
        and getattr(self.semantic_cache, "metric", "cosine") != "cosine"
        and "threshold" not in semantic_cache
    ):
        # DEFAULT_SEMANTIC_THRESHOLD (0.92) is calibrated for cosine's
        # [-1, 1] range. Falling back to it silently for e.g. "euclidean"
        # (whose scores are <= 0) would make every lookup miss forever —
        # a working feature going silently inert, not a loud failure.
        raise ConfigurationError(
            f"semantic_cache resolved to metric="
            f"{getattr(self.semantic_cache, 'metric', None)!r} but no explicit "
            "threshold= — the default (0.92) is calibrated for cosine similarity "
            "and won't make sense for this metric's score range"
        )
    self._semantic_threshold = (
        semantic_cache.get("threshold", DEFAULT_SEMANTIC_THRESHOLD)
        if isinstance(semantic_cache, dict)
        else DEFAULT_SEMANTIC_THRESHOLD
    )

    # Default pipeline: resolve context → exact cache → semantic (intent)
    # cache → provider call.
    self.pipeline = Pipeline("default")
    self.pipeline.add(
        ContextResolver(system_prompt=system_prompt, cache=self.cache)
    )
    if self.cache is not None:
        self.pipeline.add(CacheLookup(self.cache, bus=self.events))
    if self.semantic_cache is not None:
        if self.embedder is None:
            raise ConfigurationError(
                "semantic_cache requires an embedder= (config dict or async callable)"
            )
        self.pipeline.add(
            SemanticCacheLookup(
                self.semantic_cache,
                self.embedder,
                threshold=self._semantic_threshold,
                bus=self.events,
            )
        )
    if self.router is not None:
        self.pipeline.add(ProviderCall(self.router))
    self._pipelines["default"] = self.pipeline

    # A provider configure_telemetry created for us is ours to shut down
    # (flushing the final span batch); one the caller passed in is theirs.
    self._owned_tracer_provider: Any | None = (
        configure_telemetry(self, telemetry) if telemetry is not None else None
    )

aclose async

aclose() -> None

Alias for :meth:close, matching async-native naming (httpx, anyio).

Source code in src/byoai/runtime.py

async def aclose(self) -> None:
    """Alias for :meth:`close`, matching async-native naming (httpx, anyio)."""
    await self.close()

close async

close() -> None

Close every adapter the runtime owns (providers, caches, vector stores, embedder, telemetry). Prefer async with Runtime(...).

Source code in src/byoai/runtime.py

async def close(self) -> None:
    """Close every adapter the runtime owns (providers, caches, vector
    stores, embedder, telemetry). Prefer ``async with Runtime(...)``."""
    if self.router is not None:
        await self.router.close()
    if self.cache is not None:
        await self.cache.close()
    if self.vector_store is not None:
        await self.vector_store.close()
    if self.semantic_cache is not None:
        await self.semantic_cache.close()
    embedder_close = getattr(self.embedder, "close", None)
    if embedder_close is not None:
        await embedder_close()
    if self._owned_tracer_provider is not None:
        # Flush the exporter's final batch so last-window spans aren't lost.
        await asyncio.to_thread(self._owned_tracer_provider.shutdown)

on

on(event: str, handler: EventHandler) -> Any

Subscribe to lifecycle events (supports * wildcards).

Source code in src/byoai/runtime.py

def on(self, event: str, handler: EventHandler) -> Any:
    """Subscribe to lifecycle events (supports ``*`` wildcards)."""
    return self.events.on(event, handler)

stream async

stream(input: Any, *, pipeline: str | Pipeline | None = None, session_id: str | None = None, user_id: str | None = None, model: str | None = None, system_prompt: str | None = None, metadata: dict[str, Any] | None = None, provider_metadata: dict[str, Any] | None = None, filters: dict[str, Any] | None = None, **provider_options: Any) -> AsyncGenerator[StreamChunk, None]

Run the pipeline in streaming mode and yield token chunks.

The pipeline prepares the context (history, cache policy, retrieval, prompt); the terminal provider call streams. A middleware/stage that short-circuits with a response yields a single chunk.

Source code in src/byoai/runtime.py

async def stream(
    self,
    input: Any,
    *,
    pipeline: str | Pipeline | None = None,
    session_id: str | None = None,
    user_id: str | None = None,
    model: str | None = None,
    system_prompt: str | None = None,
    metadata: dict[str, Any] | None = None,
    provider_metadata: dict[str, Any] | None = None,
    filters: dict[str, Any] | None = None,
    **provider_options: Any,
) -> AsyncGenerator[StreamChunk, None]:
    """Run the pipeline in streaming mode and yield token chunks.

    The pipeline prepares the context (history, cache policy, retrieval,
    prompt); the terminal provider call streams. A middleware/stage that
    short-circuits with a response yields a single chunk.
    """
    if self.router is None:
        raise ConfigurationError("streaming requires configured providers (llm=/providers=)")
    resolved = self._resolve_pipeline(pipeline)
    ctx = self._make_context(
        input,
        resolved,
        session_id=session_id,
        user_id=user_id,
        metadata=metadata or {},
    )
    ctx.model = model
    ctx.system_prompt = system_prompt
    ctx.state[STATE_STREAMING] = True
    if filters:
        ctx.state["filters"] = filters
    if provider_metadata is not None:
        provider_options["metadata"] = provider_metadata
    if provider_options:
        ctx.state["provider_options"] = provider_options

    await self.events.emit(ev.REQUEST_RECEIVED, ctx=ctx)
    try:
        await self.middleware.execute(
            ctx, lambda c: resolved.execute(c, bus=self.events)
        )
    except Exception:
        await self.events.emit(ev.REQUEST_FAILED, ctx=ctx)
        raise

    if ctx.short_circuited and ctx.response is not None:
        yield StreamChunk(delta=ctx.response)
        yield StreamChunk(
            done=True, model=ctx.model, provider=ctx.provider,
            cached=ctx.cached, request_id=ctx.request_id,
            finish_reason=ctx.finish_reason,
        )
        await self.events.emit(ev.REQUEST_COMPLETED, ctx=ctx)
        return

    # Stages may have adjusted the options during the pipeline run.
    options = dict(ctx.state.get("provider_options", {}))
    if ctx.model:
        options["model"] = ctx.model
    parts: list[str] = []
    async for chunk in self.router.stream(ctx.messages, **options):
        if chunk.done:
            ctx.model = chunk.model or ctx.model
            ctx.provider = chunk.provider or ctx.provider
            ctx.finish_reason = chunk.finish_reason
            ctx.raw_response = chunk.raw
            if chunk.usage:
                ctx.usage.add(chunk.usage)
            # A new chunk (not the provider's raw one) so cached/request_id
            # — which the provider adapter has no knowledge of — ride the
            # final frame too, matching ExecutionResult's full result shape.
            # raw carries forward (e.g. the provider's own response id, full
            # tool_use content) so REQUEST_COMPLETED subscribers — audit
            # logging, usage recording — have the same escape hatch
            # execute() gives via ExecutionResult.raw.
            yield StreamChunk(
                done=True, model=ctx.model, provider=ctx.provider, usage=chunk.usage,
                cached=ctx.cached, request_id=ctx.request_id,
                finish_reason=ctx.finish_reason, raw=chunk.raw,
            )
        else:
            parts.append(chunk.delta)
            yield chunk
    ctx.response = "".join(parts)
    # Exact-match cache skips streaming (no STATE_CACHE_KEY set), but the
    # semantic cache stores streamed answers for future intent hits.
    await self._write_back_cache(ctx)
    await self.events.emit(ev.RESPONSE_STREAMED, ctx=ctx)
    await self.events.emit(ev.REQUEST_COMPLETED, ctx=ctx)

use

use(middleware: MiddlewareLike) -> Runtime

Add a middleware wrapping every execution. Chainable.

Source code in src/byoai/runtime.py

def use(self, middleware: MiddlewareLike) -> Runtime:
    """Add a middleware wrapping every execution. Chainable."""
    self.middleware.add(middleware)
    return self

Pipeline

byoai.Pipeline

Pipeline(name: str = 'default')

An ordered list of stages executed against one :class:RequestContext.

Stages run sequentially; a stage calling ctx.short_circuit() stops the rest. Mutate with :meth:add, :meth:remove, :meth:replace.

Source code in src/byoai/pipeline.py

def __init__(self, name: str = "default") -> None:
    self.name = name
    self._stages: list[PipelineStage] = []

add

add(stage: PipelineStage | Callable[[RequestContext], Awaitable[None]]) -> Pipeline

Append a stage. Accepts stage objects or bare async functions. Chainable.

Source code in src/byoai/pipeline.py

def add(self, stage: PipelineStage | Callable[[RequestContext], Awaitable[None]]) -> Pipeline:
    """Append a stage. Accepts stage objects or bare async functions. Chainable."""
    if not hasattr(stage, "execute"):
        stage = FunctionStage(stage)  # type: ignore[arg-type]
    self._stages.append(stage)  # type: ignore[arg-type]
    return self

remove

remove(stage_type: type | None = None, *, name: str | None = None) -> Pipeline

Remove every stage matching stage_type and/or name (both, if both given). stage_type alone matches every stage of that type — for two or more bare-function stages (every one of them is a FunctionStage once add() wraps it), pass name= too (a bare function's stage name defaults to fn.__name__) to target one specifically instead of removing all of them at once.

Source code in src/byoai/pipeline.py

def remove(self, stage_type: type | None = None, *, name: str | None = None) -> Pipeline:
    """Remove every stage matching ``stage_type`` and/or ``name`` (both, if
    both given). ``stage_type`` alone matches every stage of that type —
    for two or more bare-function stages (every one of them is a
    ``FunctionStage`` once ``add()`` wraps it), pass ``name=`` too (a bare
    function's stage name defaults to ``fn.__name__``) to target one
    specifically instead of removing all of them at once."""
    if stage_type is None and name is None:
        raise ConfigurationError("Pipeline.remove() requires stage_type and/or name=")
    self._stages = [s for s in self._stages if not self._matches(s, stage_type, name)]
    return self

replace

replace(stage_type: type | None = None, replacement: PipelineStage | None = None, *, name: str | None = None) -> Pipeline

Replace every stage matching stage_type and/or name with replacement. Same multiple-bare-function-stages caveat as :meth:remove — pass name= to target one specifically.

Source code in src/byoai/pipeline.py

def replace(
    self,
    stage_type: type | None = None,
    replacement: PipelineStage | None = None,
    *,
    name: str | None = None,
) -> Pipeline:
    """Replace every stage matching ``stage_type`` and/or ``name`` with
    ``replacement``. Same multiple-bare-function-stages caveat as
    :meth:`remove` — pass ``name=`` to target one specifically."""
    if stage_type is None and name is None:
        raise ConfigurationError("Pipeline.replace() requires stage_type and/or name=")
    if replacement is None:
        raise ConfigurationError("Pipeline.replace() requires replacement=")
    self._stages = [
        replacement if self._matches(s, stage_type, name) else s for s in self._stages
    ]
    return self

byoai.PipelineStage

Bases: Protocol

byoai.FunctionStage

FunctionStage(fn: Callable[[RequestContext], Awaitable[None]], name: str | None = None)

Adapts a bare async function into a stage.

Source code in src/byoai/pipeline.py

def __init__(
    self, fn: Callable[[RequestContext], Awaitable[None]], name: str | None = None
) -> None:
    self._fn = fn
    self.name = name or getattr(fn, "__name__", "function_stage")

Middleware

byoai.Middleware

Base class. Subclass and override __call__; plain async callables with the same (ctx, call_next) signature are also accepted by the chain.

Request context & results

byoai.RequestContext dataclass

RequestContext(input: Any, pipeline_name: str | None = None, request_id: str = (lambda: uuid.uuid4().hex)(), session_id: str | None = None, user_id: str | None = None, system_prompt: str | None = None, messages: list[Message] = list(), documents: list[Document] = list(), response: str | None = None, model: str | None = None, provider: str | None = None, finish_reason: str | None = None, raw_response: Any = None, usage: Usage = Usage(), cached: bool = False, short_circuited: bool = False, started_at: float = time.monotonic(), state: dict[str, Any] = dict(), metadata: dict[str, Any] = dict())

Carries one request through the runtime.

input is the caller's raw payload, untouched. Stages append their products to the typed slots below (messages, documents, response) or to the free-form state dict for anything stage-specific.

short_circuit

short_circuit(response: str, *, cached: bool = False) -> None

Set a final response and stop remaining pipeline stages from running.

Used by cache middleware/stages on a hit, or by guardrail middleware rejecting a request.

Source code in src/byoai/context.py

def short_circuit(self, response: str, *, cached: bool = False) -> None:
    """Set a final response and stop remaining pipeline stages from running.

    Used by cache middleware/stages on a hit, or by guardrail middleware
    rejecting a request.
    """
    self.response = response
    self.cached = cached
    self.short_circuited = True

byoai.ExecutionResult dataclass

ExecutionResult(content: str, context: RequestContext, usage: Usage = Usage(), cached: bool = False, model: str | None = None, provider: str | None = None, metadata: dict[str, Any] = dict(), finish_reason: str | None = None, raw: Any = None)

Final result of runtime.execute().

Porting from the raw anthropic/openai SDKs: content here is always a flattened str, unlike anthropic.Message.content (a list of typed content blocks) — a pure tool_use turn comes back as "", not a list to iterate. Block-level detail (tool_use, response id, prompt-cache token counts) lives on raw instead; see the docs/guides/providers.md#anthropic-tool-use-and-content-blocks guide for the per-adapter shape of raw and a worked tool-use example.

byoai.Message dataclass

Message(role: Role, content: str | list[dict[str, Any]] | None, name: str | None = None, metadata: dict[str, Any] = dict(), tool_call_id: str | None = None, tool_calls: list[dict[str, Any]] | None = None)

content is plain text, a list of provider content blocks (e.g. Anthropic tool_use/tool_result/image blocks — only the Anthropic adapters, providers/anthropic.py and providers/anthropic_cloud.py, are safe to use those with; Gemini/ OpenAI-compat expect plain strings and will error or mishandle a list), or None on an OpenAI-shaped assistant turn whose only content is a tool call (tool_calls set, mirroring OpenAI's own wire format).

tool_call_id/tool_calls are the OpenAI-compatible-family counterpart to Anthropic's content-block tool use: to send a tool result back, append the assistant's own tool-call message (role="assistant", content=None, tool_calls=[{"id": ..., "type": "function", "function": {"name": ..., "arguments": "..."}}] — the shape ExecutionResult.raw/a streamed turn's assembled tool_calls are already in) followed by one role="tool" message per call (tool_call_id=<call id>, content=<the tool's output as a string>). Anthropic instead round-trips through list-valued content blocks — see docs/guides/providers.md for both worked examples. Unused by Anthropic/Bedrock/Vertex/Gemini.

byoai.Usage dataclass

Usage(input_tokens: int = 0, output_tokens: int = 0, cost_usd: float = 0.0, cache_read_tokens: int = 0, cache_creation_tokens: int = 0)

Token/cost accounting for one or more provider calls. Additive.

byoai.StreamChunk dataclass

StreamChunk(delta: str = '', done: bool = False, model: str | None = None, provider: str | None = None, usage: Usage | None = None, raw: Any = None, cached: bool = False, request_id: str | None = None, finish_reason: str | None = None, tool_call: ToolCallDelta | None = None)

One streamed increment. delta is the new text; done marks the end.

tool_call is set instead of delta when the increment is a streamed tool-use argument fragment rather than text — the two are mutually exclusive per chunk.

cached/request_id are only ever set on the final done chunk Runtime.stream() yields (mirroring ExecutionResult), never on individual provider-level deltas.

byoai.Document dataclass

Document(id: str, content: str, metadata: dict[str, Any] = dict(), score: float | None = None, embedding: list[float] | None = None)

A retrieved vector-store document, normalized across providers.

byoai.ProviderResponse dataclass

ProviderResponse(content: str, model: str, provider: str, usage: Usage = Usage(), finish_reason: str | None = None, raw: Any = None)

Normalized non-streaming completion from any provider adapter.

Errors

byoai.ByoAIError

Bases: Exception

Base class for all runtime errors.

byoai.ConfigurationError

Bases: ByoAIError

Invalid or missing runtime configuration.

byoai.ProviderError

ProviderError(message: str, *, provider: str, status_code: int | None = None, retryable: bool = False, retry_after: float | None = None)

Bases: ByoAIError

Normalized provider failure.

Attributes:

Name	Type	Description
provider		name of the provider adapter that raised.
status_code		HTTP status from the provider, if any.
retryable		whether the router may retry this call.
retry_after		server-suggested delay in seconds, if the provider sent one.

Source code in src/byoai/errors.py

def __init__(
    self,
    message: str,
    *,
    provider: str,
    status_code: int | None = None,
    retryable: bool = False,
    retry_after: float | None = None,
) -> None:
    super().__init__(message)
    self.provider = provider
    self.status_code = status_code
    self.retryable = retryable
    self.retry_after = retry_after

byoai.AllProvidersFailedError

AllProvidersFailedError(message: str, errors: list[ProviderError])

Bases: ByoAIError

Every provider in the fallback chain failed.

Source code in src/byoai/errors.py

def __init__(self, message: str, errors: list[ProviderError]) -> None:
    super().__init__(message)
    self.errors = errors

byoai.RateLimitError

RateLimitError(message: str, *, provider: str, retry_after: float | None = None)

Bases: ProviderError

Source code in src/byoai/errors.py

def __init__(self, message: str, *, provider: str, retry_after: float | None = None) -> None:
    super().__init__(
        message,
        provider=provider,
        status_code=429,
        retryable=True,
        retry_after=retry_after,
    )

byoai.CacheError

Bases: ByoAIError

A cache adapter failed. Cache failures should generally be non-fatal.

byoai.VectorStoreError

Bases: ByoAIError

A vector store adapter failed.

byoai.FilterError

Bases: ByoAIError

An AST filter expression is malformed or unsupported by the target dialect.

byoai.MiddlewareError

Bases: ByoAIError

A middleware failed outside of normal short-circuiting.

byoai.PipelineError

PipelineError(message: str, *, stage: str | None = None)

Bases: ByoAIError

A pipeline stage failed while executing.

Source code in src/byoai/errors.py

def __init__(self, message: str, *, stage: str | None = None) -> None:
    super().__init__(message)
    self.stage = stage

byoai.PipelineNotFoundError

Bases: ByoAIError, LookupError

A named pipeline was requested but never registered.

Cache adapters

byoai.cache.base.CacheStore

Bases: Protocol

read_session async

read_session(**params: str) -> Any | None

Read existing application state through the configured pattern.

Params fill the pattern's placeholders, e.g. read_session(user_id="usr_1") with pattern app:users:{user_id}:chat_history.

Source code in src/byoai/cache/base.py

async def read_session(self, **params: str) -> Any | None:
    """Read existing application state through the configured pattern.

    Params fill the pattern's placeholders, e.g.
    ``read_session(user_id="usr_1")`` with pattern
    ``app:users:{user_id}:chat_history``.
    """
    ...

byoai.cache.memory.MemoryCache

MemoryCache(*, namespace: str = 'byoai:', default_ttl: int | None = 3600, session_reader: dict[str, str] | None = None, session_data: dict[str, Any] | None = None, max_size: int | None = None)

Source code in src/byoai/cache/memory.py

def __init__(
    self,
    *,
    namespace: str = "byoai:",
    default_ttl: int | None = 3600,
    session_reader: dict[str, str] | None = None,
    session_data: dict[str, Any] | None = None,
    max_size: int | None = None,
) -> None:
    self.namespace = namespace
    self.default_ttl = default_ttl
    self.max_size = max_size
    self._store: dict[str, tuple[Any, float | None]] = {}
    # Simulated "existing app state" for the read-only session reader.
    self._session_data = session_data or {}
    self._session_pattern = (session_reader or {}).get("pattern")

byoai.cache.redis.RedisCache

RedisCache(*, url: str = 'redis://localhost:6379', namespace: str = 'byoai:', session_reader: dict[str, str] | None = None, client: Any | None = None, default_ttl: int | None = 3600, mode: str = 'standalone', sentinels: list | None = None, service_name: str | None = None, **client_kwargs: Any)

**client_kwargs (e.g. socket_timeout, socket_connect_timeout, retry_on_timeout, health_check_interval, ssl, ssl_ca_certs) are forwarded to the underlying redis-py client when client= isn't supplied directly — see redis-py's connection docs for the full set.

Source code in src/byoai/cache/redis.py

def __init__(
    self,
    *,
    url: str = "redis://localhost:6379",
    namespace: str = "byoai:",
    session_reader: dict[str, str] | None = None,
    client: Any | None = None,
    default_ttl: int | None = 3600,
    mode: str = "standalone",
    sentinels: list | None = None,
    service_name: str | None = None,
    **client_kwargs: Any,
) -> None:
    """``**client_kwargs`` (e.g. ``socket_timeout``, ``socket_connect_timeout``,
    ``retry_on_timeout``, ``health_check_interval``, ``ssl``, ``ssl_ca_certs``)
    are forwarded to the underlying redis-py client when ``client=`` isn't
    supplied directly — see redis-py's connection docs for the full set."""
    if client is None:
        client = make_redis_client(
            url=url, mode=mode, sentinels=sentinels, service_name=service_name,
            **client_kwargs,
        )
    # Explicitly typed (not just inferred): the `client is None` branch above
    # reassigns the same declared-Optional parameter, and pyright doesn't
    # narrow a reassigned `Any | None` parameter across the branch merge —
    # without this it treats every `self._client.foo()` below as Optional.
    self._client: Any = client
    self.namespace = namespace
    self.default_ttl = default_ttl
    session_reader = session_reader or {}
    self._session_pattern: str | None = session_reader.get("pattern")
    self._session_format: str = session_reader.get("format", "json")

read_session async

read_session(**params: str) -> Any | None

Read existing app state via the configured key pattern. Never writes.

Two sequential round trips (TYPE then LRANGE/GET) — the key's shape isn't known ahead of time, and Redis has no single command that fetches "whatever this key is." ContextResolver calls this on every request that has both cache= and history reading enabled (max_history_messages > 0, the default), so this is real per-request Redis latency on top of the LLM call itself, not a one-time cost.

Source code in src/byoai/cache/redis.py

async def read_session(self, **params: str) -> Any | None:
    """Read existing app state via the configured key pattern. Never writes.

    Two sequential round trips (``TYPE`` then ``LRANGE``/``GET``) — the
    key's shape isn't known ahead of time, and Redis has no single
    command that fetches "whatever this key is." ``ContextResolver``
    calls this on every request that has both ``cache=`` and history
    reading enabled (``max_history_messages`` > 0, the default), so this
    is real per-request Redis latency on top of the LLM call itself, not
    a one-time cost.
    """
    if not self._session_pattern:
        return None
    key = self._session_pattern.format(**params)
    try:
        key_type = await self._client.type(key)
        if key_type in ("none", b"none"):
            return None
        if key_type in ("list", b"list"):
            items = await self._client.lrange(key, 0, -1)
            return [self._decode(i) for i in items]
        raw = await self._client.get(key)
    except Exception as exc:
        raise CacheError(f"redis session read failed: {exc}") from exc
    return self._decode(raw)

byoai.cache.redis.make_redis_client

make_redis_client(*, url: str = 'redis://localhost:6379', mode: str = 'standalone', sentinels: list[tuple[str, int]] | list[list] | None = None, service_name: str | None = None, **client_kwargs: Any) -> Any

Build an async Redis client for the deployment you already run.

• standalone (default) — single node or Valkey, via url.
• cluster — Redis Cluster, via url pointing at any node.
• sentinel — Sentinel-managed master: pass sentinels as [(host, port), ...] and the service_name.

Requires the redis extra: pip install byoai-runtime[redis].

Source code in src/byoai/cache/redis.py

def make_redis_client(
    *,
    url: str = "redis://localhost:6379",
    mode: str = "standalone",
    sentinels: list[tuple[str, int]] | list[list] | None = None,
    service_name: str | None = None,
    **client_kwargs: Any,
) -> Any:
    """Build an async Redis client for the deployment you already run.

    * ``standalone`` (default) — single node or Valkey, via ``url``.
    * ``cluster`` — Redis Cluster, via ``url`` pointing at any node.
    * ``sentinel`` — Sentinel-managed master: pass ``sentinels`` as
      ``[(host, port), ...]`` and the ``service_name``.

    Requires the ``redis`` extra: ``pip install byoai-runtime[redis]``.
    """
    try:
        import redis.asyncio as aioredis
    except ImportError as exc:  # pragma: no cover
        raise ConfigurationError(
            "Redis support requires the redis package: pip install 'byoai-runtime[redis]'"
        ) from exc

    if mode != "sentinel" and (sentinels or service_name):
        raise ConfigurationError(
            f"sentinels/service_name were given but mode={mode!r} — they're only used "
            "when mode='sentinel', so this would silently connect without them"
        )
    client_kwargs.setdefault("decode_responses", True)
    if mode == "standalone":
        return aioredis.from_url(url, **client_kwargs)
    if mode == "cluster":
        from redis.asyncio.cluster import RedisCluster

        return RedisCluster.from_url(url, **client_kwargs)
    if mode == "sentinel":
        if not sentinels or not service_name:
            raise ConfigurationError(
                "sentinel mode requires 'sentinels' ([(host, port), ...]) and 'service_name'"
            )
        from redis.asyncio.sentinel import Sentinel

        sentinel = Sentinel([tuple(s) for s in sentinels], **client_kwargs)
        return sentinel.master_for(service_name)
    raise ConfigurationError(
        f"unknown redis mode {mode!r} (expected standalone, cluster, or sentinel)"
    )

Semantic (intent) cache

byoai.cache.semantic.SemanticCacheStore

Bases: Protocol

find async

find(embedding: list[float], *, threshold: float) -> tuple[str, float] | None

Best cached response scoring >= threshold under the store's configured metric (cosine similarity by default), as (response, score); None on miss.

Source code in src/byoai/cache/semantic.py

async def find(
    self, embedding: list[float], *, threshold: float
) -> tuple[str, float] | None:
    """Best cached response scoring >= threshold under the store's
    configured ``metric`` (cosine similarity by default), as
    ``(response, score)``; None on miss."""
    ...

byoai.cache.semantic.MemorySemanticCache

MemorySemanticCache(*, capacity: int = 10000, ttl: int | None = 3600, metric: MetricName | SimilarityFn = 'cosine')

Ring-buffer semantic cache: fixed capacity, oldest entries evicted.

TTL is wall-clock seconds per entry (None = no expiry) — note the constructor arg here is ttl, not default_ttl like :class:~byoai.cache.memory.MemoryCache/ :class:~byoai.cache.redis.RedisCache: this store has no per-call ttl= override to be a default for (:meth:add takes no ttl= param), so there's only ever the one constructor-level value. metric selects how a query vector is scored against stored ones — a preset name or a bare callable:

• "cosine" (default) — cosine similarity, range [-1, 1]. Vectors are L2-normalized at insert and query time so a hit is a single matrix-vector product. The module docstring's threshold guidance (0.85-0.95+) assumes this metric.
• "dot" — raw inner product, no normalization. Useful when an embedding model's vector magnitude is itself meaningful.
• "euclidean" — negative squared Euclidean distance (higher = closer, unbounded range), no normalization.
• a bare callable (matrix, vector) -> scores, one score per stored row, higher = more similar — receives raw (non-normalized) vectors. Whatever range your callable produces is the range threshold gets compared against. Keep it cheap: :meth:find only offloads scoring to a worker thread past _OFFLOAD_MIN_ROWS entries, a threshold sized for the built-in presets' numpy matrix-vector product — a custom callable doing real per-row work in Python runs inline (and can stall the event loop) below that row count the same as it would above it if it's expensive, since there's no way to know a callable's cost ahead of time.

Source code in src/byoai/cache/semantic.py

def __init__(
    self,
    *,
    capacity: int = 10_000,
    ttl: int | None = 3600,
    metric: MetricName | SimilarityFn = "cosine",
) -> None:
    try:
        import numpy
    except ImportError as exc:  # pragma: no cover
        raise ConfigurationError(
            "MemorySemanticCache requires numpy: pip install 'byoai-runtime[semantic]'"
        ) from exc
    self._np = numpy
    self.capacity = capacity
    self.ttl = ttl
    self.metric = metric
    if callable(metric):
        self._normalize_vector: Callable[[Any], Any] = _identity
        self._score: SimilarityFn = metric
    else:
        self._normalize_vector, self._score = resolve_preset(
            metric,
            _METRICS,
            kind="metric",
            callable_signature="(matrix, vector) -> scores",
        )
    self._matrix: Any = None  # (capacity, dim) float32, rows normalized
    self._responses: list[str | None] = [None] * capacity
    self._expires: Any = numpy.zeros(capacity, dtype=numpy.float64)
    self._next = 0
    self._count = 0
    # Serializes writers against lookups: large lookups run in a worker
    # thread, and an add() interleaving with one could tear a row mid-read.
    self._mutex = asyncio.Lock()

add async

add(embedding: list[float], response: str, *, expires_at: float | None = None) -> None

Store an entry. expires_at (monotonic clock) overrides the store's TTL — used by shared/persistent backends replaying entries whose remaining lifetime differs from a fresh one.

Source code in src/byoai/cache/semantic.py

async def add(
    self,
    embedding: list[float],
    response: str,
    *,
    expires_at: float | None = None,
) -> None:
    """Store an entry. ``expires_at`` (monotonic clock) overrides the
    store's TTL — used by shared/persistent backends replaying entries
    whose remaining lifetime differs from a fresh one."""
    if expires_at is None:
        if self.ttl is not None and self.ttl <= 0:
            return  # a non-positive TTL means "expire immediately"
        expires_at = (
            (time.monotonic() + self.ttl) if self.ttl is not None else float("inf")
        )
    elif expires_at <= time.monotonic():
        return  # already expired
    vector = self._normalize(embedding)
    async with self._mutex:
        if self._matrix is None:
            self._matrix = self._np.zeros(
                (self.capacity, vector.shape[0]), dtype=self._np.float32
            )
        slot = self._next
        self._matrix[slot] = vector
        self._responses[slot] = response
        self._expires[slot] = expires_at
        self._next = (self._next + 1) % self.capacity
        self._count = min(self._count + 1, self.capacity)

byoai.cache.semantic.RedisSemanticCache

RedisSemanticCache(*, url: str = 'redis://localhost:6379', stream: str = 'byoai:semcache', capacity: int = 10000, ttl: int | None = 3600, metric: MetricName | SimilarityFn = 'cosine', client: Any | None = None, mode: str = 'standalone', sentinels: list | None = None, service_name: str | None = None, approximate_trim: bool = True, **client_kwargs: Any)

Shared, persistent semantic cache on an existing Redis/Valkey.

Entries live in one Redis Stream under the isolated byoai: namespace (embedding packed as base64 float32 + response + wall-clock expiry). Every worker keeps a local numpy mirror and catches up incrementally (XRANGE from its last-seen id) before each lookup — usually an empty round-trip. So intent hits are shared across processes/replicas and survive restarts, while similarity math stays local and fast.

XTRIM MAXLEN ~capacity bounds the stream; expiry is enforced at lookup time via each entry's wall-clock deadline. metric selects the similarity scoring — same presets/callable escape hatch as :class:MemorySemanticCache, which does the actual scoring here.

Requires the redis and semantic extras.

Source code in src/byoai/cache/semantic.py

def __init__(
    self,
    *,
    url: str = "redis://localhost:6379",
    stream: str = "byoai:semcache",
    capacity: int = 10_000,
    ttl: int | None = 3600,
    metric: MetricName | SimilarityFn = "cosine",
    client: Any | None = None,
    mode: str = "standalone",
    sentinels: list | None = None,
    service_name: str | None = None,
    approximate_trim: bool = True,
    **client_kwargs: Any,
) -> None:
    if client is None:
        from .redis import make_redis_client

        client = make_redis_client(
            url=url, mode=mode, sentinels=sentinels, service_name=service_name,
            **client_kwargs,
        )
    # Explicitly typed: see the matching comment in cache/redis.py.
    self._client: Any = client
    self.stream = stream
    # False = exact XTRIM MAXLEN (costlier) instead of "~" approximate
    # trimming — for deployments that need an exact capacity bound.
    self.approximate_trim = approximate_trim
    self.capacity = capacity
    self.ttl = ttl
    self.metric = metric
    self._mirror = MemorySemanticCache(capacity=capacity, ttl=ttl, metric=metric)
    self._last_id = "0-0"

Vector store adapters

byoai.vector.base.VectorStore

Bases: Protocol

byoai.vector.base.FunctionVectorStore

FunctionVectorStore(fn: Callable[..., Awaitable[list[Document]]], *, name: str | None = None)

Adapts a bare async search function into a :class:VectorStore — for a custom retrieval backend that doesn't fit a declarative vector_store={...} config. search() is the only operation this protocol has, so one function is the whole adapter — no class needed:

async def my_search(embedding: list[float], *, top_k=5, filters=None) -> list[Document]:
    ...

Runtime(vector_store=my_search)  # auto-wrapped

Source code in src/byoai/vector/base.py

def __init__(
    self,
    fn: Callable[..., Awaitable[list[Document]]],
    *,
    name: str | None = None,
) -> None:
    self._fn = fn
    self.name = name or getattr(fn, "__name__", "function_vector_store")

byoai.vector.pgvector.PgVectorStore

PgVectorStore(*, dsn: str | None = None, table: str, schema_map: dict[str, str] | None = None, metric: PgMetric = 'cosine', pool: Any | None = None, min_pool_size: int = 1, max_pool_size: int = 5, command_timeout: float | None = None, **pool_kwargs: Any)

metric must match the vector_*_ops operator class the table's index was actually built with: "cosine" (default, vector_cosine_ops), "l2" (vector_l2_ops), or "inner_product" (vector_ip_ops).

**pool_kwargs (e.g. server_settings={"statement_timeout": "..."}, ssl=..., max_inactive_connection_lifetime=...) are forwarded to asyncpg.create_pool when pool= isn't supplied directly.

Source code in src/byoai/vector/pgvector.py

def __init__(
    self,
    *,
    dsn: str | None = None,
    table: str,
    schema_map: dict[str, str] | None = None,
    metric: PgMetric = "cosine",
    pool: Any | None = None,
    min_pool_size: int = 1,
    max_pool_size: int = 5,
    command_timeout: float | None = None,
    **pool_kwargs: Any,
) -> None:
    """``metric`` must match the ``vector_*_ops`` operator class the table's index
    was actually built with: ``"cosine"`` (default, ``vector_cosine_ops``), ``"l2"``
    (``vector_l2_ops``), or ``"inner_product"`` (``vector_ip_ops``).

    ``**pool_kwargs`` (e.g. ``server_settings={"statement_timeout": "..."}``,
    ``ssl=...``, ``max_inactive_connection_lifetime=...``) are forwarded
    to ``asyncpg.create_pool`` when ``pool=`` isn't supplied directly."""
    if pool is None and dsn is None:
        raise ConfigurationError("PgVectorStore requires a dsn or an existing pool")
    # asyncpg's own parameter names for the same two knobs — accepting
    # both spellings would let **pool_kwargs silently overrule
    # min_pool_size/max_pool_size with no error. (command_timeout isn't
    # in this set: it's spelled identically in both, so passing it twice
    # is a SyntaxError at the call site, not a path into **pool_kwargs.)
    collisions = {"min_size", "max_size"} & pool_kwargs.keys()
    if collisions:
        raise ConfigurationError(
            f"use min_pool_size/max_pool_size, not {sorted(collisions)} "
            "(asyncpg's own names) — passing both would silently pick one"
        )
    self._dsn = dsn
    self._pool = pool
    self._pool_opts = {
        "min_size": min_pool_size,
        "max_size": max_pool_size,
        "command_timeout": command_timeout,
        **pool_kwargs,
    }
    # Backing fields for the table/schema_map/metric properties below,
    # set directly here (not through the properties) to avoid each
    # setter's _rebuild_query() running against a still-partially-
    # constructed instance; _rebuild_query() runs once at the end
    # instead, once all three are actually in place.
    self._table = _ident(table)
    self._schema_map = {**DEFAULT_SCHEMA_MAP, **(schema_map or {})}
    for column in self._schema_map.values():
        _ident(column)
    if metric not in _PG_OPERATORS:
        raise ConfigurationError(
            f"unknown metric {metric!r} (expected one of {sorted(_PG_OPERATORS)})"
        )
    self._metric: PgMetric = metric
    self._rebuild_query()

byoai.vector.qdrant.QdrantVectorStore

QdrantVectorStore(*, url: str = 'http://localhost:6333', collection: str, api_key: str | None = None, schema_map: dict[str, Any] | None = None, timeout: float = 30.0, client: AsyncClient | None = None, with_vectors: bool = False, score_threshold: float | None = None, search_params: dict[str, Any] | None = None)

Source code in src/byoai/vector/qdrant.py

def __init__(
    self,
    *,
    url: str = "http://localhost:6333",
    collection: str,
    api_key: str | None = None,
    schema_map: dict[str, Any] | None = None,
    timeout: float = 30.0,
    client: httpx.AsyncClient | None = None,
    with_vectors: bool = False,
    score_threshold: float | None = None,
    search_params: dict[str, Any] | None = None,
) -> None:
    self.collection = collection
    schema_map = schema_map or {}
    # payload field holding the text; None means "no content field"
    self._content_field = schema_map.get("content", "content")
    # payload field holding metadata; None means "the whole payload"
    self._metadata_field = schema_map.get("metadata", None)
    self.with_vectors = with_vectors
    # A server-side similarity floor (drop below this before top_k even
    # applies) and HNSW search params (e.g. {"hnsw_ef": 128, "exact": False}
    # to trade recall for latency) — Qdrant-specific, so exposed here
    # rather than through the cross-provider filter dialect.
    self.score_threshold = score_threshold
    self.search_params = search_params
    headers = {"api-key": api_key} if api_key else {}
    headers.setdefault("User-Agent", USER_AGENT)
    self._client = client or httpx.AsyncClient(
        base_url=url.rstrip("/"), headers=headers, timeout=timeout
    )
    self._owns_client = client is None

byoai.vector.pinecone.PineconeVectorStore

PineconeVectorStore(*, host: str | None = None, api_key: str | None = None, namespace: str = '', schema_map: dict[str, str] | None = None, timeout: float = 30.0, client: AsyncClient | None = None, include_values: bool = False, sparse_vector: dict[str, Any] | None = None)

Source code in src/byoai/vector/pinecone.py

def __init__(
    self,
    *,
    host: str | None = None,
    api_key: str | None = None,
    namespace: str = "",
    schema_map: dict[str, str] | None = None,
    timeout: float = 30.0,
    client: httpx.AsyncClient | None = None,
    include_values: bool = False,
    sparse_vector: dict[str, Any] | None = None,
) -> None:
    api_key = api_key or os.environ.get("PINECONE_API_KEY")
    # Every other adapter (Anthropic/Gemini/OpenAI-compatible) fails fast
    # here with a named env var instead of a bare TypeError from a missing
    # required kwarg — client=None still needs both to build the default
    # httpx.AsyncClient, so this check only backs off when client= is
    # already fully constructed (same escape hatch those adapters use).
    if client is None and (not host or not api_key):
        raise ConfigurationError(
            "PineconeVectorStore requires 'host' and 'api_key' (or $PINECONE_API_KEY) "
            "unless a pre-built client= is supplied"
        )
    self.namespace = namespace
    # metadata field holding the document text (Pinecone stores text in metadata)
    self._content_field = (schema_map or {}).get("content", "content")
    self.include_values = include_values
    # A fixed sparse component for hybrid dense+sparse search
    # ({"indices": [...], "values": [...]}); per-query sparse vectors
    # aren't supported by the fixed VectorStore.search() signature.
    self.sparse_vector = sparse_vector
    if client is None:
        # Narrowed by the ConfigurationError check above: client is None
        # implies host/api_key are both non-empty here.
        assert host is not None and api_key is not None
        client = httpx.AsyncClient(
            base_url=host.rstrip("/"),
            headers={"Api-Key": api_key, "User-Agent": USER_AGENT},
            timeout=timeout,
        )
        self._owns_client = True
    else:
        self._owns_client = False
    self._client = client

Provider adapters

byoai.providers.base.LLMProvider

Bases: Protocol

byoai.providers.base.FunctionProvider

FunctionProvider(fn: Callable[..., Awaitable[str | ProviderResponse]], *, name: str | None = None, model: str = '', stream_fn: Callable[..., AsyncIterator[str | StreamChunk]] | None = None)

Adapts a bare async function into an :class:LLMProvider — for a custom or gateway-wrapped backend that doesn't fit a declarative llm={...} config. Mirrors Pipeline's FunctionStage and the embedder= callable pattern: give the runtime one function, not a class.

async def my_gateway(messages: list[Message], **options) -> str:
    response = await my_existing_client.create(...)
    return response.text

Runtime(providers=[my_gateway])  # auto-wrapped — no class needed

fn may return a plain str (model/usage/finish_reason default to empty/zero) or a full :class:ProviderResponse when you want those tracked. providers=[...] passed to :class:ProviderRouter auto-wraps any bare callable this way; you rarely need to construct this directly.

Source code in src/byoai/providers/base.py

def __init__(
    self,
    fn: Callable[..., Awaitable[str | ProviderResponse]],
    *,
    name: str | None = None,
    model: str = "",
    stream_fn: Callable[..., AsyncIterator[str | StreamChunk]] | None = None,
) -> None:
    self._fn = fn
    self.name = name or getattr(fn, "__name__", "function_provider")
    self.model = model
    self._stream_fn = stream_fn

byoai.providers.router.ProviderRouter

ProviderRouter(providers: Sequence[LLMProvider | Callable[..., Any]], *, retry_policy: RetryPolicy | None = None, selection: SelectionName | SelectionFn = 'ordered', event_bus: EventBus | None = None)

Tries providers in the order selection picks for each call: retryable failures back off and retry up to retry_policy.max_retries, then routing falls through to the next provider in that order; when every provider fails, :class:AllProvidersFailedError carries the accumulated errors.

selection — a preset name or a bare callable:

• "ordered" (default) — always try providers in the order given; identical to the router's original fixed-primary/fallback behavior.
• "round_robin" — rotates which provider goes first each call, so load spreads across providers instead of always preferring the first; a failure still falls through the rest in ring order, so nothing is skipped, only reprioritized.
• a bare callable (providers) -> providers, returning the providers to try, in order, for this call — e.g. weighted selection. A callable that also filters (e.g. dropping providers it considers unhealthy) excludes them entirely for that call, by its own choice; it is not a pure reprioritization like the two presets above.

providers accepts LLMProvider instances or bare async functions — a callable without a complete attribute is auto-wrapped in :class:FunctionProvider, same as Pipeline.add() auto-wraps a bare function into a FunctionStage.

Source code in src/byoai/providers/router.py

def __init__(
    self,
    providers: Sequence[LLMProvider | Callable[..., Any]],
    *,
    retry_policy: RetryPolicy | None = None,
    selection: SelectionName | SelectionFn = "ordered",
    event_bus: EventBus | None = None,
) -> None:
    """``providers`` accepts ``LLMProvider`` instances or bare async
    functions — a callable without a ``complete`` attribute is
    auto-wrapped in :class:`FunctionProvider`, same as ``Pipeline.add()``
    auto-wraps a bare function into a ``FunctionStage``."""
    if not providers:
        raise ValueError("ProviderRouter requires at least one provider")
    wrapped = [
        p if hasattr(p, "complete") else FunctionProvider(p)  # type: ignore[arg-type]
        for p in providers
    ]
    self.providers = cast("list[LLMProvider]", wrapped)
    self.retry_policy = retry_policy or RetryPolicy()
    self.selection = selection
    if callable(selection):
        self._select: SelectionFn = selection
    else:
        # _SELECTIONS stores zero-arg factories (not ready-to-use
        # selectors) so "round_robin" gets its own fresh, per-router
        # _RoundRobinSelection instance rather than one shared globally.
        factory = resolve_preset(
            selection,
            _SELECTIONS,
            kind="selection",
            callable_signature="(providers) -> providers",
        )
        self._select = factory()
    self._bus = event_bus

stream async

stream(messages: list[Message], **options: Any) -> AsyncIterator[StreamChunk]

Stream from the first provider that starts successfully.

Fallback happens only if a provider fails before yielding any content; once tokens have been emitted downstream, a mid-stream failure is raised as-is (the transport already sent partial output).

Source code in src/byoai/providers/router.py

async def stream(
    self, messages: list[Message], **options: Any
) -> AsyncIterator[StreamChunk]:
    """Stream from the first provider that starts successfully.

    Fallback happens only if a provider fails *before yielding any content*;
    once tokens have been emitted downstream, a mid-stream failure is raised
    as-is (the transport already sent partial output).
    """
    errors: list[ProviderError] = []
    for provider in self._selected_providers():
        attempt = 0
        while True:
            await self._emit(
                ev.PROVIDER_STARTED, provider=provider.name, model=provider.model
            )
            yielded = False
            try:
                async for chunk in provider.stream(messages, **options):
                    if chunk.done:
                        await self._emit(
                            ev.PROVIDER_COMPLETED,
                            provider=provider.name,
                            model=chunk.model,
                            usage=chunk.usage,
                        )
                    else:
                        yielded = True
                    yield chunk
                return
            except ProviderError as exc:
                errors.append(exc)
                await self._emit(ev.PROVIDER_FAILED, provider=provider.name, error=str(exc))
                if yielded:
                    raise
                if not exc.retryable or attempt >= self.retry_policy.max_retries:
                    break
                await asyncio.sleep(self.retry_policy.delay(attempt, exc.retry_after))
                attempt += 1
    raise AllProvidersFailedError(
        "; ".join(str(e) for e in errors) or "all providers failed", errors
    )

byoai.providers.router.RetryPolicy dataclass

RetryPolicy(max_retries: int = 2, base_delay: float = 0.5, max_delay: float = 10.0, jitter: float = 0.25)

Retry/backoff knobs for :class:ProviderRouter: exponential backoff from base_delay capped at max_delay, with proportional jitter; a server-provided Retry-After wins (still capped at max_delay).

byoai.providers.openai_compat.OpenAICompatProvider

OpenAICompatProvider(*, model: str, api_key: str | None = None, base_url: str = 'https://api.openai.com/v1', name: str = 'openai', timeout: float = 60.0, default_headers: dict[str, str] | None = None, default_params: dict[str, str] | None = None, client: AsyncClient | None = None, chat_path: str = '/chat/completions', retryable_status: frozenset[int] | set[int] | None = None)

Source code in src/byoai/providers/openai_compat.py

def __init__(
    self,
    *,
    model: str,
    api_key: str | None = None,
    base_url: str = "https://api.openai.com/v1",
    name: str = "openai",
    timeout: float = 60.0,
    default_headers: dict[str, str] | None = None,
    default_params: dict[str, str] | None = None,
    client: httpx.AsyncClient | None = None,
    chat_path: str = "/chat/completions",
    retryable_status: frozenset[int] | set[int] | None = None,
) -> None:
    self.name = name
    self.model = model
    self._chat_path = chat_path
    self._retryable_status = (
        frozenset(retryable_status) if retryable_status is not None
        else DEFAULT_RETRYABLE_STATUS
    )
    self._client, self._owns_client = build_openai_client(
        api_key=api_key, base_url=base_url, timeout=timeout,
        default_headers=default_headers, default_params=default_params,
        client=client,
    )

byoai.providers.anthropic.AnthropicProvider

AnthropicProvider(*, model: str, api_key: str | None = None, base_url: str = 'https://api.anthropic.com', name: str = 'anthropic', timeout: float = 60.0, max_tokens: int = 4096, client: AsyncClient | None = None, api_version: str = DEFAULT_API_VERSION, default_headers: dict[str, str] | None = None, retryable_status: frozenset[int] | set[int] | None = None, messages_path: str = '/v1/messages', cache_system: bool = False)

Source code in src/byoai/providers/anthropic.py

def __init__(
    self,
    *,
    model: str,
    api_key: str | None = None,
    base_url: str = "https://api.anthropic.com",
    name: str = "anthropic",
    timeout: float = 60.0,
    max_tokens: int = 4096,
    client: httpx.AsyncClient | None = None,
    api_version: str = DEFAULT_API_VERSION,
    default_headers: dict[str, str] | None = None,
    retryable_status: frozenset[int] | set[int] | None = None,
    messages_path: str = "/v1/messages",
    cache_system: bool = False,
) -> None:
    self.name = name
    self.model = model
    self.max_tokens = max_tokens
    self._messages_path = messages_path
    # When True, a plain-string system prompt is wrapped in Anthropic's
    # cache_control ephemeral block so repeated calls with the same
    # prompt hit the server-side prompt cache. No effect on a system
    # message whose content is already a list of content blocks (the
    # caller built their own — see build_anthropic_system_field).
    self.cache_system = cache_system
    self._retryable_status = (
        frozenset(retryable_status) if retryable_status is not None
        else DEFAULT_RETRYABLE_STATUS_ANTHROPIC
    )
    self._owns_client = client is None
    if client is None:
        api_key = api_key or os.environ.get("ANTHROPIC_API_KEY")
        if not api_key and not has_auth_header(default_headers, "x-api-key", "authorization"):
            # Fail fast at construction rather than sending a credential-less
            # request and surfacing a confusing 401 at request time. A
            # default_headers= carrying its own auth header (this adapter's,
            # however capitalized, or a generic Authorization) disables this
            # check: a gateway may authenticate under a different scheme —
            # but an unrelated header (e.g. a tracing header) must not.
            raise ConfigurationError(
                "AnthropicProvider needs an API key: pass api_key= or set "
                "the ANTHROPIC_API_KEY environment variable (or supply your "
                "own auth via default_headers=)"
            )
        headers = {"anthropic-version": api_version, "User-Agent": USER_AGENT}
        if api_key:
            headers["x-api-key"] = api_key
        headers.update(default_headers or {})
        client = httpx.AsyncClient(
            base_url=base_url.rstrip("/"), headers=headers, timeout=timeout,
        )
    self._client = client

byoai.providers.anthropic_cloud.AnthropicBedrockProvider

AnthropicBedrockProvider(*, model: str, name: str = 'bedrock', max_tokens: int = 4096, aws_region: str | None = None, aws_access_key: str | None = None, aws_secret_key: str | None = None, aws_session_token: str | None = None, aws_profile: str | None = None, default_headers: dict[str, str] | None = None, retryable_status: frozenset[int] | set[int] | None = None, client: Any | None = None, cache_system: bool = False)

Bases: _AnthropicSDKProviderBase

Anthropic models via AWS Bedrock.

Only aws_region is required here; credentials come from the standard AWS chain (env vars, ~/.aws, or an instance/task role) unless passed explicitly. Declarative: llm={"provider": "bedrock", "model": "...", "aws_region": "us-east-1"}.

Source code in src/byoai/providers/anthropic_cloud.py

def __init__(
    self,
    *,
    model: str,
    name: str = "bedrock",
    max_tokens: int = 4096,
    aws_region: str | None = None,
    aws_access_key: str | None = None,
    aws_secret_key: str | None = None,
    aws_session_token: str | None = None,
    aws_profile: str | None = None,
    default_headers: dict[str, str] | None = None,
    retryable_status: frozenset[int] | set[int] | None = None,
    client: Any | None = None,
    cache_system: bool = False,
) -> None:
    self.name = name
    self.model = model
    self.max_tokens = max_tokens
    self.cache_system = cache_system
    self._retryable_status = (
        frozenset(retryable_status)
        if retryable_status is not None
        else DEFAULT_RETRYABLE_STATUS_ANTHROPIC
    )
    if client is not None:
        self._client = client
        self._owns_client = False
        return
    try:
        # pyright flags this as a private-module re-export, but it's the
        # SDK's own documented public import path (confirmed at runtime).
        from anthropic import AsyncAnthropicBedrock  # pyright: ignore[reportPrivateImportUsage]
    except ImportError as exc:
        raise ConfigurationError(
            "AnthropicBedrockProvider requires the anthropic[bedrock] package: "
            "pip install 'byoai-runtime[bedrock]'"
        ) from exc
    region = aws_region or os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION")
    if not region:
        raise ConfigurationError(
            "AnthropicBedrockProvider requires aws_region (or $AWS_REGION / "
            "$AWS_DEFAULT_REGION); AWS credentials themselves come from the standard "
            "chain (env vars, ~/.aws, or an instance/task role) unless passed explicitly."
        )
    kwargs: dict[str, Any] = {"aws_region": region}
    if aws_access_key:
        kwargs["aws_access_key"] = aws_access_key
    if aws_secret_key:
        kwargs["aws_secret_key"] = aws_secret_key
    if aws_session_token:
        kwargs["aws_session_token"] = aws_session_token
    if aws_profile:
        kwargs["aws_profile"] = aws_profile
    if default_headers:
        kwargs["default_headers"] = default_headers
    self._client = AsyncAnthropicBedrock(**kwargs)
    self._owns_client = True

byoai.providers.anthropic_cloud.AnthropicVertexProvider

AnthropicVertexProvider(*, model: str, name: str = 'vertex', max_tokens: int = 4096, project_id: str | None = None, region: str | None = None, access_token: str | None = None, credentials: Any | None = None, default_headers: dict[str, str] | None = None, retryable_status: frozenset[int] | set[int] | None = None, client: Any | None = None, cache_system: bool = False)

Bases: _AnthropicSDKProviderBase

Anthropic models via Google Vertex AI.

project_id and region are required; credentials come from Application Default Credentials unless access_token/credentials is passed explicitly. Declarative: llm={"provider": "vertex", "model": "...", "project_id": "...", "region": "us-east5"}.

Source code in src/byoai/providers/anthropic_cloud.py

def __init__(
    self,
    *,
    model: str,
    name: str = "vertex",
    max_tokens: int = 4096,
    project_id: str | None = None,
    region: str | None = None,
    access_token: str | None = None,
    credentials: Any | None = None,
    default_headers: dict[str, str] | None = None,
    retryable_status: frozenset[int] | set[int] | None = None,
    client: Any | None = None,
    cache_system: bool = False,
) -> None:
    self.name = name
    self.model = model
    self.max_tokens = max_tokens
    self.cache_system = cache_system
    self._retryable_status = (
        frozenset(retryable_status)
        if retryable_status is not None
        else DEFAULT_RETRYABLE_STATUS_ANTHROPIC
    )
    if client is not None:
        self._client = client
        self._owns_client = False
        return
    try:
        # See the matching note in AnthropicBedrockProvider above.
        from anthropic import AsyncAnthropicVertex  # pyright: ignore[reportPrivateImportUsage]
    except ImportError as exc:
        raise ConfigurationError(
            "AnthropicVertexProvider requires the anthropic[vertex] package: "
            "pip install 'byoai-runtime[vertex]'"
        ) from exc
    resolved_project = (
        project_id
        or os.environ.get("ANTHROPIC_VERTEX_PROJECT_ID")
        or os.environ.get("GOOGLE_CLOUD_PROJECT")
    )
    resolved_region = (
        region or os.environ.get("ANTHROPIC_VERTEX_REGION") or os.environ.get("CLOUD_ML_REGION")
    )
    if not (resolved_project and resolved_region):
        raise ConfigurationError(
            "AnthropicVertexProvider requires project_id and region (or "
            "$ANTHROPIC_VERTEX_PROJECT_ID/$GOOGLE_CLOUD_PROJECT and "
            "$ANTHROPIC_VERTEX_REGION/$CLOUD_ML_REGION); GCP credentials themselves "
            "come from Application Default Credentials unless access_token/credentials "
            "is passed explicitly."
        )
    kwargs: dict[str, Any] = {"project_id": resolved_project, "region": resolved_region}
    if access_token:
        kwargs["access_token"] = access_token
    if credentials is not None:
        kwargs["credentials"] = credentials
    if default_headers:
        kwargs["default_headers"] = default_headers
    self._client = AsyncAnthropicVertex(**kwargs)
    self._owns_client = True

byoai.providers.gemini.GeminiProvider

GeminiProvider(*, model: str, api_key: str | None = None, base_url: str = 'https://generativelanguage.googleapis.com/v1beta', name: str = 'gemini', timeout: float = 60.0, client: AsyncClient | None = None, default_headers: dict[str, str] | None = None, retryable_status: frozenset[int] | set[int] | None = None)

Source code in src/byoai/providers/gemini.py

def __init__(
    self,
    *,
    model: str,
    api_key: str | None = None,
    base_url: str = "https://generativelanguage.googleapis.com/v1beta",
    name: str = "gemini",
    timeout: float = 60.0,
    client: httpx.AsyncClient | None = None,
    default_headers: dict[str, str] | None = None,
    retryable_status: frozenset[int] | set[int] | None = None,
) -> None:
    self.name = name
    self.model = model
    self._retryable_status = (
        frozenset(retryable_status) if retryable_status is not None
        else DEFAULT_RETRYABLE_STATUS
    )
    self._owns_client = client is None
    if client is None:
        api_key = (
            api_key
            or os.environ.get("GEMINI_API_KEY")
            or os.environ.get("GOOGLE_API_KEY")
        )
        if not api_key and not has_auth_header(
            default_headers, "x-goog-api-key", "authorization"
        ):
            # Fail fast at construction rather than sending a credential-less
            # request and surfacing a confusing 401 at request time. A
            # default_headers= carrying its own auth header (this adapter's,
            # however capitalized, or a generic Authorization) disables this
            # check: a gateway may authenticate under a different scheme —
            # but an unrelated header (e.g. a tracing header) must not.
            raise ConfigurationError(
                "GeminiProvider needs an API key: pass api_key= or set the "
                "GEMINI_API_KEY (or GOOGLE_API_KEY) environment variable "
                "(or supply your own auth via default_headers=)"
            )
        headers = {"User-Agent": USER_AGENT}
        if api_key:
            headers["x-goog-api-key"] = api_key
        headers.update(default_headers or {})
        client = httpx.AsyncClient(
            base_url=base_url.rstrip("/"), headers=headers, timeout=timeout,
        )
    self._client = client

byoai.providers.embeddings.OpenAICompatEmbedder

OpenAICompatEmbedder(*, model: str, api_key: str | None = None, base_url: str = 'https://api.openai.com/v1', name: str = 'openai', timeout: float = 30.0, default_headers: dict[str, str] | None = None, client: AsyncClient | None = None, embeddings_path: str = '/embeddings', retryable_status: frozenset[int] | set[int] | None = None, max_batch_size: int | None = None)

Source code in src/byoai/providers/embeddings.py

def __init__(
    self,
    *,
    model: str,
    api_key: str | None = None,
    base_url: str = "https://api.openai.com/v1",
    name: str = "openai",
    timeout: float = 30.0,
    default_headers: dict[str, str] | None = None,
    client: httpx.AsyncClient | None = None,
    embeddings_path: str = "/embeddings",
    retryable_status: frozenset[int] | set[int] | None = None,
    max_batch_size: int | None = None,
) -> None:
    self.name = name
    self.model = model
    self._embeddings_path = embeddings_path
    self._retryable_status = (
        frozenset(retryable_status) if retryable_status is not None
        else DEFAULT_RETRYABLE_STATUS
    )
    # e.g. OpenAI caps /embeddings at 2048 inputs per call; large batch
    # ingestion jobs are chunked transparently rather than erroring.
    self.max_batch_size = max_batch_size
    self._client, self._owns_client = build_openai_client(
        api_key=api_key, base_url=base_url, timeout=timeout,
        default_headers=default_headers, client=client,
    )

Pipeline stages

byoai.stages.ContextResolver

ContextResolver(*, system_prompt: str | None = None, cache: CacheStore | None = None, session_params: Callable[[RequestContext], dict[str, str]] | None = None, max_history_messages: int = 20)

Normalize ctx.input into ctx.messages.

Accepts

• str — becomes a single user message
• {"messages": [{"role": ..., "content": ...}, ...]}
• {"query"|"input"|"prompt": str, ...}
• list[Message] / list of role-content dicts

Optionally prepends a system prompt and existing session history read (read-only) from the cache adapter's session reader.

Source code in src/byoai/stages.py

def __init__(
    self,
    *,
    system_prompt: str | None = None,
    cache: CacheStore | None = None,
    session_params: Callable[[RequestContext], dict[str, str]] | None = None,
    max_history_messages: int = 20,
) -> None:
    self.system_prompt = system_prompt
    self.cache = cache
    self.session_params = session_params
    self.max_history_messages = max_history_messages

byoai.stages.CacheLookup

CacheLookup(cache: CacheStore, *, bus: EventBus | None = None, extra_fingerprint: Callable[[RequestContext], Any] | None = None)

Exact-match response cache. Short-circuits the pipeline on a hit.

The cache key fingerprints the normalized messages, model, pipeline, provider options (temperature, top_p, ...) and retrieval filters, so requests that differ only in those fields never collide on the same entry. The runtime writes the response back (with the cache's own default_ttl) after a successful non-streamed, non-cached execution.

Source code in src/byoai/stages.py

def __init__(
    self,
    cache: CacheStore,
    *,
    bus: EventBus | None = None,
    extra_fingerprint: Callable[[RequestContext], Any] | None = None,
) -> None:
    self.cache = cache
    self._bus = bus
    # Hook for apps needing extra key dimensions (e.g. a tenant id from
    # ctx.state) without subclassing this stage.
    self.extra_fingerprint = extra_fingerprint

byoai.stages.SemanticCacheLookup

SemanticCacheLookup(store: Any, embedder: Embedder, *, threshold: float = DEFAULT_SEMANTIC_THRESHOLD, bus: EventBus | None = None)

Intent cache: short-circuit when a similar (not identical) query was already answered. Runs after the exact-match cache — exact hits are cheaper (no embedding call). On a miss, the query embedding is kept on the context so the runtime can store the eventual response for future intent hits.

Streaming requests participate too: a hit streams back as a single chunk.

Source code in src/byoai/stages.py

def __init__(
    self,
    store: Any,  # SemanticCacheStore
    embedder: Embedder,
    *,
    threshold: float = DEFAULT_SEMANTIC_THRESHOLD,
    bus: EventBus | None = None,
) -> None:
    self.store = store
    self.embedder = embedder
    self.threshold = threshold
    self._bus = bus

byoai.stages.VectorRetrieve

VectorRetrieve(store: VectorStore, embedder: Embedder, *, top_k: int = 5, filters: dict[str, Any] | None = None, bus: EventBus | None = None, format_document: Callable[[Document], str] | None = None, context_header: str = 'Relevant context retrieved for this request:', insert_at: Callable[[RequestContext], int] | None = None)

Retrieve documents from an existing vector store for the last user message.

embedder is supplied by the application (it owns embedding model choice); ByoAI executes the retrieval. Filters come from ctx.state['filters'] or the constructor default, in the unified AST dialect.

Source code in src/byoai/stages.py

def __init__(
    self,
    store: VectorStore,
    embedder: Embedder,
    *,
    top_k: int = 5,
    filters: dict[str, Any] | None = None,
    bus: EventBus | None = None,
    format_document: Callable[[Document], str] | None = None,
    context_header: str = "Relevant context retrieved for this request:",
    insert_at: Callable[[RequestContext], int] | None = None,
) -> None:
    self.store = store
    self.embedder = embedder
    self.top_k = top_k
    self.filters = filters
    self._bus = bus
    # Customize the RAG prompt wrapper (citation format, instructions to
    # the model) without subclassing this stage.
    self.format_document = format_document or (lambda d: f"[{d.id}] {d.content}")
    self.context_header = context_header
    # Default: just before the last message. Override to e.g. always
    # append at the end, or merge into an existing system message.
    self.insert_at = insert_at or (lambda ctx: max(len(ctx.messages) - 1, 0))

byoai.stages.ProviderCall

ProviderCall(router: ProviderRouter, **default_options: Any)

Terminal stage: call the provider router with the built messages.

In streaming mode (ctx.state[STATE_STREAMING]) this stage is a no-op — the runtime streams from the router itself after the pipeline finishes preparing the context.

Source code in src/byoai/stages.py

def __init__(self, router: ProviderRouter, **default_options: Any) -> None:
    self.router = router
    self.default_options = default_options

Telemetry (OpenTelemetry)

byoai.telemetry.otel.instrument

instrument(runtime: Runtime, *, tracer_provider: Any | None = None) -> Runtime

Attach tracing to a runtime: request spans, per-stage child spans, and provider lifecycle span events.

Source code in src/byoai/telemetry/otel.py

def instrument(runtime: Runtime, *, tracer_provider: Any | None = None) -> Runtime:
    """Attach tracing to a runtime: request spans, per-stage child spans, and
    provider lifecycle span events."""
    middleware = OpenTelemetryMiddleware(tracer_provider)
    runtime.use(middleware)
    tracer = middleware.tracer

    def on_stage_started(event: str, payload: dict[str, Any]) -> None:
        ctx: RequestContext = payload["ctx"]
        span = tracer.start_span(f"byoai.stage.{payload['stage']}")
        # Make the stage span the active context for the stage's duration so
        # spans created inside the stage (e.g. instrumented HTTP clients)
        # parent to the stage, not the request. Stages run sequentially in the
        # same task between the started/completed events, so attach/detach
        # pairs nest correctly.
        token = otel_context.attach(trace.set_span_in_context(span))
        ctx.state.setdefault(_STATE_KEY, {})[payload["stage"]] = (span, token)

    def on_stage_completed(event: str, payload: dict[str, Any]) -> None:
        ctx: RequestContext = payload["ctx"]
        entry = ctx.state.get(_STATE_KEY, {}).pop(payload["stage"], None)
        if entry is not None:
            span, token = entry
            otel_context.detach(token)
            span.end()

    def on_provider_event(event: str, payload: dict[str, Any]) -> None:
        # GenAI semantic-convention keys, matching the span attributes set by
        # OpenTelemetryMiddleware._finalize.
        attributes: dict[str, Any] = {"gen_ai.system": payload.get("provider", "")}
        if payload.get("model"):
            key = (
                "gen_ai.response.model"
                if event == ev.PROVIDER_COMPLETED
                else "gen_ai.request.model"
            )
            attributes[key] = payload["model"]
        if payload.get("error"):
            attributes["error.message"] = payload["error"]
        usage = payload.get("usage")
        if usage is not None:
            attributes["gen_ai.usage.input_tokens"] = usage.input_tokens
            attributes["gen_ai.usage.output_tokens"] = usage.output_tokens
        trace.get_current_span().add_event(event, attributes)

    def on_cache_event(event: str, payload: dict[str, Any]) -> None:
        trace.get_current_span().add_event(event)

    runtime.on(ev.STAGE_STARTED, on_stage_started)
    runtime.on(ev.STAGE_COMPLETED, on_stage_completed)
    runtime.on("provider.*", on_provider_event)
    runtime.on("cache.*", on_cache_event)
    return runtime

byoai.telemetry.otel.configure_otlp

configure_otlp(*, endpoint: str, service_name: str = 'byoai-runtime', headers: dict[str, str] | None = None, protocol: str = 'grpc', timeout: float | None = None, compression: str | None = None, resource_attributes: dict[str, Any] | None = None, max_queue_size: int | None = None, schedule_delay_millis: int | None = None, max_export_batch_size: int | None = None, export_timeout_millis: int | None = None) -> Any

Create an SDK TracerProvider exporting OTLP to an existing collector (Grafana Tempo, Datadog agent, Honeycomb, Jaeger, ...). Returns the provider — pass it to :func:instrument.

protocol is "grpc" (default, port 4317) or "http"/"http/protobuf" (port 4318) — many collectors behind corporate ingress only allow the latter. compression is "gzip" or None. The max_queue_size/ schedule_delay_millis/max_export_batch_size/export_timeout_millis knobs tune the batch processor; unset ones use the OTel SDK's defaults. resource_attributes adds to (not replaces) service.name — e.g. {"service.version": "1.2.0", "deployment.environment": "prod"}.

Source code in src/byoai/telemetry/otel.py

def configure_otlp(
    *,
    endpoint: str,
    service_name: str = "byoai-runtime",
    headers: dict[str, str] | None = None,
    protocol: str = "grpc",
    timeout: float | None = None,
    compression: str | None = None,
    resource_attributes: dict[str, Any] | None = None,
    max_queue_size: int | None = None,
    schedule_delay_millis: int | None = None,
    max_export_batch_size: int | None = None,
    export_timeout_millis: int | None = None,
) -> Any:
    """Create an SDK TracerProvider exporting OTLP to an existing collector
    (Grafana Tempo, Datadog agent, Honeycomb, Jaeger, ...). Returns the
    provider — pass it to :func:`instrument`.

    ``protocol`` is ``"grpc"`` (default, port 4317) or ``"http"``/``"http/protobuf"``
    (port 4318) — many collectors behind corporate ingress only allow the
    latter. ``compression`` is ``"gzip"`` or ``None``. The ``max_queue_size``/
    ``schedule_delay_millis``/``max_export_batch_size``/``export_timeout_millis``
    knobs tune the batch processor; unset ones use the OTel SDK's defaults.
    ``resource_attributes`` adds to (not replaces) ``service.name`` — e.g.
    ``{"service.version": "1.2.0", "deployment.environment": "prod"}``.
    """
    if protocol not in ("grpc", "http", "http/protobuf"):
        raise ValueError(f"unknown OTLP protocol {protocol!r} (expected 'grpc' or 'http')")
    if compression not in (None, "gzip"):
        raise ValueError(f"unknown OTLP compression {compression!r} (expected 'gzip' or None)")

    try:
        from opentelemetry.sdk.resources import Resource
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.trace.export import BatchSpanProcessor

        if protocol == "grpc":
            from grpc import Compression as _Compression
            from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
        else:
            from opentelemetry.exporter.otlp.proto.http import Compression as _Compression
            from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    except ImportError as exc:
        raise ImportError(
            "OTLP export requires the SDK + exporter: pip install 'byoai-runtime[otel]'"
        ) from exc

    # _Compression/OTLPSpanExporter resolve to one of two distinct classes
    # (grpc vs http) depending on `protocol`, which pyright can't narrow from a
    # runtime string check — it merges both branches' imports into a union and
    # then can't satisfy either constructor's specific `compression` enum type.
    # Both branches are otherwise identical calls, so widen to Any rather than
    # duplicate the exporter construction per branch.
    compression_value: Any = _Compression.Gzip if compression == "gzip" else None
    exporter = OTLPSpanExporter(
        endpoint=endpoint, headers=headers, timeout=timeout, compression=compression_value
    )
    resource = Resource.create({"service.name": service_name, **(resource_attributes or {})})
    provider = TracerProvider(resource=resource)
    # Built from a plain dict[str, int], so pyright checks its value type
    # against every keyword BatchSpanProcessor accepts when unpacked —
    # including the unrelated `meter_provider: MeterProvider | None` — rather
    # than only the four keys this dict can actually contain.
    batch_kwargs: dict[str, Any] = {
        k: v
        for k, v in {
            "max_queue_size": max_queue_size,
            "schedule_delay_millis": schedule_delay_millis,
            "max_export_batch_size": max_export_batch_size,
            "export_timeout_millis": export_timeout_millis,
        }.items()
        if v is not None
    }
    provider.add_span_processor(BatchSpanProcessor(exporter, **batch_kwargs))
    return provider

byoai.telemetry.otel.OpenTelemetryMiddleware

OpenTelemetryMiddleware(tracer_provider: Any | None = None)

Bases: Middleware

Wraps every execution in a byoai.execute span.

Source code in src/byoai/telemetry/otel.py

def __init__(self, tracer_provider: Any | None = None) -> None:
    self._tracer = trace.get_tracer("byoai", tracer_provider=tracer_provider)

Background workers

byoai.workers.RuntimeWorker

RuntimeWorker(runtime: Runtime, queue: JobQueue, *, concurrency: int = 10, shutdown_timeout: float | None = None)

Consume jobs and execute them through the runtime, concurrency at a time. Failed jobs get an {"error": ...} result and are still acked (dead-lettering/retry policy belongs to the queue, not the worker).

Source code in src/byoai/workers.py

def __init__(
    self,
    runtime: Runtime,
    queue: JobQueue,
    *,
    concurrency: int = 10,
    shutdown_timeout: float | None = None,
) -> None:
    self.runtime = runtime
    self.queue = queue
    self.concurrency = concurrency
    # Caps how long stop()/run() waits for in-flight jobs to finish
    # draining; None (default) waits indefinitely. A stuck job otherwise
    # blocks graceful shutdown forever.
    self.shutdown_timeout = shutdown_timeout
    self._stopping = asyncio.Event()
    self._in_flight: set[asyncio.Task] = set()
    self.processed = 0
    self.failed = 0
    # Distinct from `failed` (a job the runtime itself couldn't answer,
    # still reported back through push_result/ack normally): errors
    # counts a job whose *result delivery* failed (push_result/ack
    # raised, e.g. a queue-backend blip) — the runtime answered fine but
    # the caller/queue never found out, so it needs separate visibility.
    self.errors = 0

run async

run(*, until_idle: bool = False, poll_timeout: float = 0.5) -> None

Consume jobs; drains in-flight work on exit.

Default mode runs until :meth:stop is called. With until_idle=True it returns once the queue stays empty and nothing is in flight (batch/test runs).

Source code in src/byoai/workers.py

async def run(self, *, until_idle: bool = False, poll_timeout: float = 0.5) -> None:
    """Consume jobs; drains in-flight work on exit.

    Default mode runs until :meth:`stop` is called. With ``until_idle=True``
    it returns once the queue stays empty and nothing is in flight
    (batch/test runs).
    """
    semaphore = asyncio.Semaphore(self.concurrency)
    # At full concurrency, semaphore.acquire() alone can block past a
    # stop() call until some in-flight job happens to finish — race it
    # against the stop signal so shutdown_timeout can actually take
    # effect instead of waiting on an already-slow/stuck job. One
    # long-lived stop_task is reused across iterations (it only resolves
    # once, when stop() fires) rather than spun up fresh per job.
    stop_task = asyncio.ensure_future(self._stopping.wait())
    try:
        while not self._stopping.is_set():
            acquire_task = asyncio.ensure_future(semaphore.acquire())
            await asyncio.wait(
                {acquire_task, stop_task}, return_when=asyncio.FIRST_COMPLETED
            )
            if stop_task.done():
                if acquire_task.done():
                    semaphore.release()  # acquired right as we were stopping; give it back
                else:
                    acquire_task.cancel()
                break
            job = await self.queue.pop(timeout=poll_timeout)
            if job is None:
                semaphore.release()
                if until_idle:
                    if not self._in_flight:
                        return
                    await self._drain()
                continue
            task = asyncio.create_task(self._process(job, semaphore))
            self._in_flight.add(task)
            task.add_done_callback(self._in_flight.discard)
    finally:
        stop_task.cancel()
    await self._drain()

run_until_idle async

run_until_idle() -> None

Convenience for batch/test runs: consume until the queue stays empty.

Source code in src/byoai/workers.py

async def run_until_idle(self) -> None:
    """Convenience for batch/test runs: consume until the queue stays empty."""
    await self.run(until_idle=True, poll_timeout=0.1)

byoai.workers.JobQueue

Bases: Protocol

pop async

pop(timeout: float = 1.0) -> Job | None

Next job, or None if none arrived within timeout seconds.

Source code in src/byoai/workers.py

async def pop(self, timeout: float = 1.0) -> Job | None:
    """Next job, or None if none arrived within ``timeout`` seconds."""
    ...

byoai.workers.Job dataclass

Job(payload: dict[str, Any], id: str = (lambda: uuid.uuid4().hex)(), delivery_tag: Any = None)

byoai.workers.MemoryJobQueue

MemoryJobQueue(*, maxsize: int = 0)

In-process queue for dev/tests. Same contract as RedisStreamQueue.

maxsize (default 0 = unbounded, matching asyncio.Queue) bounds memory when publishers can outrun a slow worker fleet; publish() then backpressures by awaiting free space instead of growing forever.

Source code in src/byoai/workers.py

def __init__(self, *, maxsize: int = 0) -> None:
    self._queue: asyncio.Queue[Job] = asyncio.Queue(maxsize=maxsize)
    self._results: dict[str, dict[str, Any]] = {}
    self.acked: list[str] = []

byoai.workers.RedisStreamQueue

RedisStreamQueue(*, url: str = 'redis://localhost:6379', stream: str = 'byoai:jobs', group: str = 'byoai-workers', consumer: str | None = None, result_prefix: str = 'byoai:result:', result_ttl: int = 3600, prefetch: int = 16, client: Any | None = None, mode: str = 'standalone', sentinels: list | None = None, service_name: str | None = None, maxlen: int | None = None, approximate_trim: bool = True, start_id: str = '0', **client_kwargs: Any)

Jobs on a Redis Stream with a consumer group; results as byoai:- namespaced keys with TTL. At-least-once delivery: entries are XACKed only after the result is stored.

Requires the redis extra: pip install byoai-runtime[redis].

maxlen caps the jobs stream (unbounded by default) so an idle or crashed worker fleet doesn't let publishers grow it forever. start_id is the consumer group's initial read position — "0" (default) replays the whole existing stream for a fresh group; "$" starts from only new entries, for attaching a new worker fleet to a pre-existing, already-large stream without a backlog replay. **client_kwargs are forwarded to the redis-py client.

Source code in src/byoai/workers.py

def __init__(
    self,
    *,
    url: str = "redis://localhost:6379",
    stream: str = "byoai:jobs",
    group: str = "byoai-workers",
    consumer: str | None = None,
    result_prefix: str = "byoai:result:",
    result_ttl: int = 3600,
    prefetch: int = 16,
    client: Any | None = None,
    mode: str = "standalone",
    sentinels: list | None = None,
    service_name: str | None = None,
    maxlen: int | None = None,
    approximate_trim: bool = True,
    start_id: str = "0",
    **client_kwargs: Any,
) -> None:
    """``maxlen`` caps the jobs stream (unbounded by default) so an idle or
    crashed worker fleet doesn't let publishers grow it forever.
    ``start_id`` is the consumer group's initial read position — ``"0"``
    (default) replays the whole existing stream for a fresh group;
    ``"$"`` starts from only new entries, for attaching a new worker
    fleet to a pre-existing, already-large stream without a backlog
    replay. ``**client_kwargs`` are forwarded to the redis-py client."""
    if client is None:
        from .cache.redis import make_redis_client

        client = make_redis_client(
            url=url, mode=mode, sentinels=sentinels, service_name=service_name,
            **client_kwargs,
        )
    # Explicitly typed: see the matching comment in cache/redis.py.
    self._client: Any = client
    self.stream = stream
    self.group = group
    self.consumer = consumer or f"worker-{uuid.uuid4().hex[:8]}"
    self.result_prefix = result_prefix
    self.result_ttl = result_ttl
    self.maxlen = maxlen
    self.approximate_trim = approximate_trim
    self.start_id = start_id
    # Batch up to `prefetch` entries per XREADGROUP round-trip; pop()
    # serves from the local buffer so throughput isn't capped at one
    # network round-trip per job.
    self.prefetch = max(1, prefetch)
    self._buffer: list[Job] = []
    self._group_ready = False

FastAPI integration

byoai.integrations.fastapi.attach

attach(app: FastAPI, runtime: Runtime) -> Runtime

Bind the runtime to an app: stores it on app.state and registers a shutdown hook that closes provider/cache/vector connections.

Works alongside an app's existing lifespan/startup handlers — it only adds an on_shutdown callback rather than replacing the lifespan.

Source code in src/byoai/integrations/fastapi.py

def attach(app: FastAPI, runtime: Runtime) -> Runtime:
    """Bind the runtime to an app: stores it on ``app.state`` and registers a
    shutdown hook that closes provider/cache/vector connections.

    Works alongside an app's existing lifespan/startup handlers — it only adds
    an ``on_shutdown`` callback rather than replacing the lifespan.
    """
    setattr(app.state, _STATE_ATTR, runtime)
    # add_event_handler is gone from the FastAPI/Starlette app class itself in
    # current versions but survives (deprecated) on FastAPI's own APIRouter —
    # hasattr-based duck typing across that version split, so `target` can't
    # be given a real static type here.
    target: Any = app if hasattr(app, "add_event_handler") else app.router
    target.add_event_handler("shutdown", runtime.close)
    return runtime

byoai.integrations.fastapi.get_runtime

get_runtime(request: Request) -> Runtime

FastAPI dependency: runtime: Runtime = Depends(get_runtime).

Also works with a WebSocket passed directly (not through Depends) since both expose .app.state — useful for fetching the runtime inside a websocket route before calling :func:serve_websocket. Typed as Request rather than Request | WebSocket because FastAPI's dependency-injection machinery cannot build a response field for a Union here; callers using the websocket form should pass it through as-is (it works at runtime) or cast it for their own type-checking.

Source code in src/byoai/integrations/fastapi.py

def get_runtime(request: Request) -> Runtime:
    """FastAPI dependency: ``runtime: Runtime = Depends(get_runtime)``.

    Also works with a ``WebSocket`` passed directly (not through ``Depends``)
    since both expose ``.app.state`` — useful for fetching the runtime inside
    a websocket route before calling :func:`serve_websocket`. Typed as
    ``Request`` rather than ``Request | WebSocket`` because FastAPI's
    dependency-injection machinery cannot build a response field for a Union
    here; callers using the websocket form should pass it through as-is (it
    works at runtime) or ``cast`` it for their own type-checking.
    """
    runtime = getattr(request.app.state, _STATE_ATTR, None)
    if runtime is None:
        raise ConfigurationError(
            "no ByoAI runtime attached to this app — call "
            "byoai.integrations.fastapi.attach(app, runtime) at startup"
        )
    return runtime

byoai.integrations.fastapi.stream_response

stream_response(runtime: Runtime, input: Any, *, media_type: str = 'text/event-stream', headers: dict[str, str] | None = None, **execute_kwargs: Any) -> StreamingResponse

SSE response streaming runtime.stream() chunks.

Emits data: {"delta": "..."} events per token batch and a final data: {"done": true, "usage": {...}} event. headers defaults to disabling proxy buffering (Cache-Control: no-cache, X-Accel-Buffering: no) — pass {} to omit them entirely.

Source code in src/byoai/integrations/fastapi.py

def stream_response(
    runtime: Runtime,
    input: Any,
    *,
    media_type: str = "text/event-stream",
    headers: dict[str, str] | None = None,
    **execute_kwargs: Any,
) -> StreamingResponse:
    """SSE response streaming ``runtime.stream()`` chunks.

    Emits ``data: {"delta": "..."}`` events per token batch and a final
    ``data: {"done": true, "usage": {...}}`` event. ``headers`` defaults to
    disabling proxy buffering (``Cache-Control: no-cache``,
    ``X-Accel-Buffering: no``) — pass ``{}`` to omit them entirely.
    """
    effective_headers = (
        {"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
        if headers is None
        else headers
    )

    async def event_source():
        try:
            async for chunk in runtime.stream(input, **execute_kwargs):
                if has_content(chunk):
                    yield f"data: {json.dumps(chunk_to_dict(chunk))}\n\n"
        except ByoAIError as exc:
            # Headers are already on the wire; a torn connection would leave the
            # client guessing. Emit the same terminal error event as
            # transport.sse_stream so all transports fail identically.
            yield f"data: {json.dumps({'error': str(exc), 'done': True})}\n\n"

    return StreamingResponse(event_source(), media_type=media_type, headers=effective_headers)

byoai.integrations.fastapi.serve_websocket async

serve_websocket(runtime: Runtime, websocket: WebSocket) -> None

Serve the shared WebSocket dialect on an accepted-or-new connection.

Each client message is one JSON payload (see byoai.transport); the response is a stream of JSON frames — {"delta": ...} per token batch, then {"done": true, "usage": {...}}. Use inside your own route::

@app.websocket("/ws")
async def ws(websocket: WebSocket, rt: Runtime = Depends(get_runtime)):
    await serve_websocket(rt, websocket)

Source code in src/byoai/integrations/fastapi.py

async def serve_websocket(runtime: Runtime, websocket: WebSocket) -> None:
    """Serve the shared WebSocket dialect on an accepted-or-new connection.

    Each client message is one JSON payload (see ``byoai.transport``); the
    response is a stream of JSON frames — ``{"delta": ...}`` per token batch,
    then ``{"done": true, "usage": {...}}``. Use inside your own route::

        @app.websocket("/ws")
        async def ws(websocket: WebSocket, rt: Runtime = Depends(get_runtime)):
            await serve_websocket(rt, websocket)
    """
    if websocket.client_state.name == "CONNECTING":
        await websocket.accept()
    try:
        while True:
            raw = await websocket.receive_text()
            async for frame in ws_reply(runtime, raw):
                await websocket.send_text(frame)
    except WebSocketDisconnect:
        pass

Flask integration

byoai.integrations.flask.attach

attach(app: Flask, runtime: Runtime) -> Runtime

Bind runtime to app via app.extensions (Flask's own extension-state slot) and start its background bridge thread.

Idempotent: calling this again on an already-attached app (a duplicate call, or Werkzeug's debug-mode reloader re-executing module code) returns the existing runtime instead of starting a second bridge thread — the runtime passed on that second call is unused and gets closed immediately (its provider(s) already opened real connections in __init__, so leaving it open would leak them for the rest of the process). Registers atexit cleanup — WSGI has no lifespan-shutdown protocol like ASGI's, so this is the best available hook; it won't fire on SIGKILL, but does on gunicorn's graceful SIGTERM path.

Source code in src/byoai/integrations/flask.py

def attach(app: Flask, runtime: Runtime) -> Runtime:
    """Bind ``runtime`` to ``app`` via ``app.extensions`` (Flask's own
    extension-state slot) and start its background bridge thread.

    Idempotent: calling this again on an already-attached app (a duplicate
    call, or Werkzeug's debug-mode reloader re-executing module code) returns
    the existing runtime instead of starting a second bridge thread — the
    ``runtime`` passed on that second call is unused and gets closed
    immediately (its provider(s) already opened real connections in
    ``__init__``, so leaving it open would leak them for the rest of the
    process). Registers ``atexit`` cleanup — WSGI has no lifespan-shutdown
    protocol like ASGI's, so this is the best available hook; it won't fire
    on SIGKILL, but does on gunicorn's graceful SIGTERM path.
    """
    existing: _FlaskBridge | None = app.extensions.get(_STATE_ATTR)
    if existing is not None:
        if runtime is not existing.runtime:
            try:
                existing.run(runtime.aclose())
            except Exception:
                logger.warning(
                    "byoai Flask bridge: failed to close a redundant Runtime "
                    "passed to a duplicate attach() call",
                    exc_info=True,
                )
        return existing.runtime
    bridge = _FlaskBridge(runtime)
    app.extensions[_STATE_ATTR] = bridge
    atexit.register(bridge.close)
    return runtime

byoai.integrations.flask.get_runtime

get_runtime(app: Flask | None = None) -> Runtime

The attached Runtime — defaults to flask.current_app; pass app= explicitly when calling outside a request/app context (e.g. a background job wrapped in with app.app_context(): ...).

Source code in src/byoai/integrations/flask.py

def get_runtime(app: Flask | None = None) -> Runtime:
    """The attached ``Runtime`` — defaults to ``flask.current_app``; pass
    ``app=`` explicitly when calling outside a request/app context (e.g. a
    background job wrapped in ``with app.app_context(): ...``)."""
    return _bridge(app).runtime

byoai.integrations.flask.execute

execute(input: Any, *, app: Flask | None = None, **kwargs: Any) -> ExecutionResult

Sync wrapper for runtime.execute() — blocks the calling (Flask request) thread until the result is ready.

Source code in src/byoai/integrations/flask.py

def execute(input: Any, *, app: Flask | None = None, **kwargs: Any) -> ExecutionResult:
    """Sync wrapper for ``runtime.execute()`` — blocks the calling (Flask
    request) thread until the result is ready."""
    bridge = _bridge(app)
    try:
        return bridge.run(bridge.runtime.execute(input, **kwargs))
    except ConcurrentCancelledError as exc:
        # run() re-raises this as-is (not wrapped in a ByoAIError) when a
        # future gets cancelled for a reason unrelated to the bridge
        # closing — the same race stream_response() already handles for
        # the streaming path (see its event_source()). A typical Flask view
        # built on this helper only expects `except ByoAIError:`, so a raw
        # concurrent.futures.CancelledError escaping here surfaces as an
        # unhandled 500 instead of a normal, catchable runtime error.
        # str(CancelledError()) is always "" (it never carries a message),
        # so fall back to a description instead of an empty one.
        raise ByoAIError(str(exc) or "request cancelled") from exc

byoai.integrations.flask.stream_response

stream_response(input: Any, *, app: Flask | None = None, media_type: str = 'text/event-stream', headers: dict[str, str] | None = None, **execute_kwargs: Any) -> Response

SSE Response streaming runtime.stream() chunks — same frame shape as the FastAPI/Robyn integrations (via transport.chunk_to_dict). headers defaults to disabling proxy buffering; pass {} to omit.

Source code in src/byoai/integrations/flask.py

def stream_response(
    input: Any,
    *,
    app: Flask | None = None,
    media_type: str = "text/event-stream",
    headers: dict[str, str] | None = None,
    **execute_kwargs: Any,
) -> Response:
    """SSE ``Response`` streaming ``runtime.stream()`` chunks — same frame
    shape as the FastAPI/Robyn integrations (via ``transport.chunk_to_dict``).
    ``headers`` defaults to disabling proxy buffering; pass ``{}`` to omit.
    """
    bridge = _bridge(app)
    effective_headers = (
        {"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
        if headers is None
        else headers
    )

    def event_source() -> Iterator[str]:
        agen = bridge.runtime.stream(input, **execute_kwargs)
        try:
            for chunk in bridge.run_stream(agen):
                if has_content(chunk):
                    yield f"data: {json.dumps(chunk_to_dict(chunk))}\n\n"
        except (ByoAIError, ConcurrentCancelledError) as exc:
            # Headers are already on the wire; emit the same terminal error
            # event as the other transports rather than tearing the
            # connection down. ConcurrentCancelledError: run() re-raises
            # this as-is (not wrapped in a ByoAIError) when a future gets
            # cancelled for a reason unrelated to the bridge closing — a
            # ByoAIError-only catch here let it escape unhandled instead of
            # ending the stream cleanly. str(CancelledError()) is always ""
            # (it never carries a message), so that case specifically falls
            # back to a description rather than an empty, undiagnosable
            # error — but only that case: a genuine ByoAIError that happens
            # to carry no message (e.g. a bring-your-own-function provider
            # raising ByoAIError() bare) must not also get mislabeled as a
            # cancellation that never happened, misleading any client-side
            # handling that branches on that specific message.
            if isinstance(exc, ConcurrentCancelledError):
                message = str(exc) or "request cancelled"
            else:
                message = str(exc) or type(exc).__name__
            yield f"data: {json.dumps({'error': message, 'done': True})}\n\n"

    # stream_with_context is required: Flask tears down the request/app
    # context as soon as the view returns unless the generator is wrapped.
    return Response(
        stream_with_context(event_source()), mimetype=media_type, headers=effective_headers
    )

Robyn integration

byoai.integrations.robyn.attach

attach(app: Robyn, runtime: Runtime, *, prefix: str = '/byoai', stream_media_type: str = 'text/event-stream', stream_headers: dict[str, str] | None = None) -> Runtime

Register ByoAI routes on an existing Robyn app and bind lifecycle.

Adds POST {prefix}/execute, POST {prefix}/stream (SSE) and a {prefix}/ws WebSocket, plus a shutdown handler closing the runtime's provider/cache/vector connections. stream_headers defaults to disabling proxy buffering (Cache-Control: no-cache, X-Accel-Buffering: no) — pass {} to omit them entirely.

Source code in src/byoai/integrations/robyn.py

def attach(
    app: Robyn,
    runtime: Runtime,
    *,
    prefix: str = "/byoai",
    stream_media_type: str = "text/event-stream",
    stream_headers: dict[str, str] | None = None,
) -> Runtime:
    """Register ByoAI routes on an existing Robyn app and bind lifecycle.

    Adds ``POST {prefix}/execute``, ``POST {prefix}/stream`` (SSE) and a
    ``{prefix}/ws`` WebSocket, plus a shutdown handler closing the runtime's
    provider/cache/vector connections. ``stream_headers`` defaults to
    disabling proxy buffering (``Cache-Control: no-cache``,
    ``X-Accel-Buffering: no``) — pass ``{}`` to omit them entirely.
    """
    effective_stream_headers = (
        {"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
        if stream_headers is None
        else stream_headers
    )

    @app.post(f"{prefix}/execute")
    async def _execute(request):  # Robyn injects its Request
        try:
            payload = request.json()
        except Exception:
            return _error_response(400, "request body must be JSON")
        try:
            # Validate eagerly so a malformed payload is a 400, distinct from
            # execution-time failures.
            parse_payload(payload)
        except ByoAIError as exc:
            return _error_response(400, str(exc))
        try:
            result = await execute_payload(runtime, payload)
        except ByoAIError as exc:
            return _error_response(_error_status(exc), str(exc), headers=_error_headers(exc))
        return jsonify(result)

    @app.post(f"{prefix}/stream")
    async def _stream(request):
        try:
            payload = request.json()
        except Exception:
            return _error_response(400, "request body must be JSON")
        try:
            # Validate before streaming starts so bad requests get a real
            # status code; mid-stream errors become SSE error events instead.
            parse_payload(payload)
        except ByoAIError as exc:
            return _error_response(400, str(exc))
        return StreamingResponse(
            sse_stream(runtime, payload),
            media_type=stream_media_type,
            # Robyn's StreamingResponse expects its native Headers type — a
            # plain dict is truthy so `headers or Headers({})` keeps the dict
            # as-is, and Robyn's own SSE-default-header code then calls
            # `.set()` on it, which a plain dict doesn't have.
            headers=Headers(effective_stream_headers),
        )

    @app.websocket(f"{prefix}/ws")
    async def _ws(websocket):
        # One JSON payload per message; frames streamed back per token batch.
        # Dialect (frames/errors) is shared via transport.ws_reply.
        try:
            while True:
                raw = await websocket.receive_text()
                async for frame in ws_reply(runtime, raw):
                    await websocket.send_text(frame)
        except Exception as exc:  # ordinary client disconnects must not raise
            if not _is_disconnect(exc):
                raise

    async def _close() -> None:
        await runtime.close()

    app.shutdown_handler(_close)
    return runtime

byoai.integrations.robyn.create_app

create_app(runtime: Runtime, *, prefix: str = '/byoai', healthz_path: str | None = '/healthz', **attach_kwargs: Any) -> Robyn

A standalone Robyn app serving the runtime.

Registers GET {healthz_path} by default; pass healthz_path=None to skip it (e.g. if your host app already defines that route).

Source code in src/byoai/integrations/robyn.py

def create_app(
    runtime: Runtime,
    *,
    prefix: str = "/byoai",
    healthz_path: str | None = "/healthz",
    **attach_kwargs: Any,
) -> Robyn:
    """A standalone Robyn app serving the runtime.

    Registers ``GET {healthz_path}`` by default; pass ``healthz_path=None``
    to skip it (e.g. if your host app already defines that route).
    """
    app = Robyn(__file__)
    attach(app, runtime, prefix=prefix, **attach_kwargs)

    if healthz_path is not None:
        @app.get(healthz_path)
        async def _healthz():
            return jsonify({"ok": True})

    return app

MCP integration

byoai.integrations.mcp.create_server

create_server(runtime: Runtime, *, name: str = 'byoai-runtime', tool_name: str = 'execute', description: str | None = None, stream_tool_name: str | None = 'execute_stream', stream_description: str | None = None, **server_kwargs: Any) -> Any

Build an MCP server exposing runtime as one or two tools.

Both tools accept the same payload shape as every other transport (input, optional pipeline, session_id, user_id, model, filters) and return the same result dict as POST /execute elsewhere.

• tool_name (default "execute") — one request, one response.
• stream_tool_name (default "execute_stream"; pass None to disable) — streams token deltas as MCP progress notifications (visible to clients that render them live) while still returning the same full result dict at the end, so non-streaming-aware clients work unchanged. Progress notifications require a live client session; if the tool is invoked without one (e.g. a local call_tool() in a test), reporting is skipped and the full result is still returned correctly.

**server_kwargs are forwarded to the underlying MCP server constructor (instructions, version, debug, auth, ...) for anything this wrapper doesn't already surface.

Source code in src/byoai/integrations/mcp.py

def create_server(
    runtime: Runtime,
    *,
    name: str = "byoai-runtime",
    tool_name: str = "execute",
    description: str | None = None,
    stream_tool_name: str | None = "execute_stream",
    stream_description: str | None = None,
    **server_kwargs: Any,
) -> Any:
    """Build an MCP server exposing ``runtime`` as one or two tools.

    Both tools accept the same payload shape as every other transport
    (``input``, optional ``pipeline``, ``session_id``, ``user_id``, ``model``,
    ``filters``) and return the same result dict as ``POST /execute`` elsewhere.

    * ``tool_name`` (default ``"execute"``) — one request, one response.
    * ``stream_tool_name`` (default ``"execute_stream"``; pass ``None`` to
      disable) — streams token deltas as MCP progress notifications (visible
      to clients that render them live) while still returning the same full
      result dict at the end, so non-streaming-aware clients work unchanged.
      Progress notifications require a live client session; if the tool is
      invoked without one (e.g. a local `call_tool()` in a test), reporting
      is skipped and the full result is still returned correctly.

    ``**server_kwargs`` are forwarded to the underlying MCP server
    constructor (``instructions``, ``version``, ``debug``, ``auth``, ...) for
    anything this wrapper doesn't already surface.
    """
    server = _MCPServerCls(name, **server_kwargs)

    @server.tool(
        name=tool_name,
        description=description
        or "Execute a request through the ByoAI runtime (routing, caching, "
        "retries, cost tracking) and return the response.",
    )
    async def execute(
        input: str,
        pipeline: str | None = None,
        session_id: str | None = None,
        user_id: str | None = None,
        model: str | None = None,
        filters: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        payload = _payload_from_args(input, pipeline, session_id, user_id, model, filters)
        try:
            return await execute_payload(runtime, payload)
        except ByoAIError as exc:
            # Surfaced to the MCP client as a tool error, not a transport crash.
            return {"error": str(exc), "error_type": type(exc).__name__}

    if stream_tool_name is not None:

        @server.tool(
            name=stream_tool_name,
            description=stream_description
            or "Like the execute tool, but streams token deltas as progress "
            "notifications while still returning the full response at the end.",
        )
        async def execute_stream(
            input: str,
            ctx: Context,
            pipeline: str | None = None,
            session_id: str | None = None,
            user_id: str | None = None,
            model: str | None = None,
            filters: dict[str, Any] | None = None,
        ) -> dict[str, Any]:
            payload = _payload_from_args(input, pipeline, session_id, user_id, model, filters)
            parts: list[str] = []
            chars_sent = 0
            final: dict[str, Any] = {}
            try:
                async for frame in stream_frames(runtime, payload):
                    if frame.get("done"):
                        final = frame
                        continue
                    delta = frame.get("delta", "")
                    if not delta:
                        continue
                    parts.append(delta)
                    chars_sent += len(delta)
                    try:
                        await ctx.report_progress(chars_sent, message=delta)
                    except ValueError:
                        pass  # no live client session (e.g. local call_tool()) — non-fatal
            except ByoAIError as exc:
                return {"error": str(exc), "error_type": type(exc).__name__}
            # Same key set as the non-streaming `execute` tool / POST /execute,
            # so a caller reading e.g. result["provider"] doesn't need to
            # special-case the streaming tool.
            return {
                "content": "".join(parts),
                "cached": final.get("cached", False),
                "model": final.get("model"),
                "provider": final.get("provider"),
                "usage": final.get(
                    "usage", {"input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0}
                ),
                "request_id": final.get("request_id"),
            }

    return server

byoai.integrations.mcp.attach

attach(app: Starlette, runtime: Runtime, *, path: str = '/mcp', name: str = 'byoai-runtime', **create_kwargs: Any) -> Any

Mount an MCP server into an existing Starlette/FastAPI app at path (streamable HTTP transport).

Two things app.mount() alone doesn't give you, both handled here:

• The MCP sub-app has its own ASGI lifespan (it starts the streamable-HTTP session manager's task group); Starlette never cascades a parent app's lifespan into a mounted sub-app, so without this the session manager never starts and every request 500s. We enter/exit that lifespan manually via the host app's startup/shutdown hooks.
• server.streamable_http_app() registers its own internal route at streamable_http_path — mounting that app again at path would double the prefix (path + path). We register the sub-app's internal route at "/" so path is applied exactly once, by app.mount().

Requires the host app to support add_event_handler (FastAPI's router, or Starlette apps not yet on the lifespan-only contract) so we have somewhere to hook the sub-app's startup/shutdown — raises :class:ConfigurationError immediately if it doesn't, rather than mounting an endpoint that would 500 on every request. Plain modern Starlette apps (lifespan= only) aren't supported by this helper; wire the MCP server's lifespan into your own instead.

Source code in src/byoai/integrations/mcp.py

def attach(
    app: Starlette,
    runtime: Runtime,
    *,
    path: str = "/mcp",
    name: str = "byoai-runtime",
    **create_kwargs: Any,
) -> Any:
    """Mount an MCP server into an existing Starlette/FastAPI app at ``path``
    (streamable HTTP transport).

    Two things ``app.mount()`` alone doesn't give you, both handled here:

    * The MCP sub-app has its own ASGI lifespan (it starts the streamable-HTTP
      session manager's task group); Starlette never cascades a parent app's
      lifespan into a mounted sub-app, so without this the session manager
      never starts and every request 500s. We enter/exit that lifespan
      manually via the host app's startup/shutdown hooks.
    * ``server.streamable_http_app()`` registers its own internal route at
      ``streamable_http_path`` — mounting *that* app again at ``path`` would
      double the prefix (``path`` + ``path``). We register the sub-app's
      internal route at ``"/"`` so ``path`` is applied exactly once, by
      ``app.mount()``.

    Requires the host app to support ``add_event_handler`` (FastAPI's
    router, or Starlette apps not yet on the lifespan-only contract) so we
    have somewhere to hook the sub-app's startup/shutdown — raises
    :class:`ConfigurationError` immediately if it doesn't, rather than
    mounting an endpoint that would 500 on every request. Plain modern
    Starlette apps (``lifespan=`` only) aren't supported by this helper;
    wire the MCP server's lifespan into your own instead.
    """
    from ..errors import ConfigurationError

    server = create_server(runtime, name=name, **create_kwargs)
    mcp_app = server.streamable_http_app(streamable_http_path="/")
    app.mount(path, mcp_app)

    # add_event_handler is gone from the Starlette/FastAPI app class itself in
    # current versions but survives (deprecated) on FastAPI's own APIRouter —
    # hasattr-based duck typing across that version split, so `target` can't
    # be given a real static type here.
    target: Any = next(
        (t for t in (app, getattr(app, "router", None)) if hasattr(t, "add_event_handler")),
        None,
    )
    if target is None:
        raise ConfigurationError(
            "byoai.integrations.mcp.attach() requires an app with "
            "add_event_handler (FastAPI, or Starlette with the on_event-style "
            "lifecycle) to start the MCP session manager's lifespan — plain "
            "modern Starlette apps using lifespan= only aren't supported here. "
            "Enter mcp_app.router.lifespan_context(mcp_app) in your own "
            "lifespan instead, where mcp_app = server.streamable_http_app()."
        )

    lifespan_cm = mcp_app.router.lifespan_context(mcp_app)

    async def _start_mcp() -> None:
        await lifespan_cm.__aenter__()

    async def _stop_mcp() -> None:
        await lifespan_cm.__aexit__(None, None, None)
        await runtime.close()

    target.add_event_handler("startup", _start_mcp)
    target.add_event_handler("shutdown", _stop_mcp)
    return server

byoai.integrations.mcp.create_app

create_app(runtime: Runtime, *, name: str = 'byoai-runtime', **create_kwargs: Any)

A standalone ASGI app serving the MCP tool over streamable HTTP.

Source code in src/byoai/integrations/mcp.py

def create_app(runtime: Runtime, *, name: str = "byoai-runtime", **create_kwargs: Any):
    """A standalone ASGI app serving the MCP tool over streamable HTTP."""
    server = create_server(runtime, name=name, **create_kwargs)
    return server.streamable_http_app()