AI Providers¶

AIProvider ¶

Bases: ABC

AI model provider for generating responses.

name `property` ¶

name

Provider name (e.g. 'anthropic', 'openai').

supports_vision `property` ¶

supports_vision

Whether this provider can process images.

supports_streaming `property` ¶

supports_streaming

Whether this provider supports streaming token generation.

supports_structured_streaming `property` ¶

supports_structured_streaming

Whether this provider supports structured streaming with tool calls.

model_name `abstractmethod` `property` ¶

model_name

Model identifier (e.g. 'claude-sonnet-4-20250514', 'gpt-4o').

generate `abstractmethod` `async` ¶

generate(context)

Generate an AI response from the given context.

Parameters:

Name	Type	Description	Default
`context`	`AIContext`	Conversation context including messages, system prompt, temperature, and target channel capabilities.	required

Returns:

Type	Description
`AIResponse`	The AI response with content, usage stats, and optional
`AIResponse`	tasks/observations.

generate_stream `async` ¶

generate_stream(context)

Yield text deltas as they arrive. Override for streaming providers.

generate_structured_stream `async` ¶

generate_structured_stream(context)

Yield structured events (thinking deltas, text deltas, tool calls, done).

Default implementation wraps generate() so every provider works without changes. Override for true streaming support.

close `async` ¶

close()

Release resources. Override in subclasses that hold connections.

AIContext ¶

Bases: BaseModel

Context passed to AI provider for generation.

AIMessage ¶

Bases: BaseModel

A message in the AI conversation context.

AITextPart ¶

Bases: BaseModel

Text part of a multimodal message.

AIImagePart ¶

Bases: BaseModel

Image part of a multimodal message.

AITool ¶

Bases: BaseModel

Tool definition for function calling.

AIToolCall ¶

Bases: BaseModel

A tool call from the AI response.

AIResponse ¶

Bases: BaseModel

Response from an AI provider.

MockAIProvider ¶

MockAIProvider(responses=None, *, vision=False, ai_responses=None, streaming=False)

Bases: AIProvider

Round-robin response provider for tests.

generate_stream `async` ¶

generate_stream(context)

Yield text from generate() as a single delta.

generate_structured_stream `async` ¶

generate_structured_stream(context)

Yield structured events from generate() result.

Per-Room AI Configuration¶

AI channel settings can be overridden per-room using binding metadata:

# Default AI channel
ai = AIChannel("ai", provider=anthropic, system_prompt="Default assistant")
kit.register_channel(ai)

# Override per room
kit.attach_channel(room_id, "ai", metadata={
    "system_prompt": "You are a customer support agent for Acme Corp.",
    "temperature": 0.3,  # More deterministic
    "max_tokens": 2048,
})

Silent Observer Pattern (Meeting Notes)¶

# Attach AI as note-taker
kit.attach_channel(meeting_room_id, "ai", metadata={
    "system_prompt": """You are a meeting note-taker.
    Listen to the conversation silently.
    When someone says 'meeting ended', compile and send a summary.""",
})

# Mute so AI listens but doesn't respond
await kit.mute(meeting_room_id, "ai")

# Later, unmute to let AI send summary
await kit.unmute(meeting_room_id, "ai")

Tools/Function Calling¶

Tools can be passed via binding metadata for function calling:

kit.attach_channel(room_id, "ai", metadata={
    "tools": [
        {
            "name": "search_knowledge_base",
            "description": "Search the company knowledge base",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"]
            }
        }
    ]
})

Tool calls are returned in AIResponse.tool_calls:

response = await provider.generate(context)
for tool_call in response.tool_calls:
    print(f"Tool: {tool_call.name}, Args: {tool_call.arguments}")

Gemini Provider¶

GeminiAIProvider ¶

GeminiAIProvider(config)

Bases: AIProvider

AI provider using the Google Gemini API.

supports_vision `property` ¶

supports_vision

All Gemini models support vision.

generate_structured_stream `async` ¶

generate_structured_stream(context)

Yield structured events from the Gemini streaming API.

generate `async` ¶

generate(context)

Generate by consuming the structured stream.

generate_stream `async` ¶

generate_stream(context)

Yield text deltas as they arrive from the Gemini API.

close `async` ¶

close()

Release the genai client reference.

GeminiConfig ¶

Bases: BaseModel

Google Gemini AI provider configuration.

Usage¶

from roomkit import GeminiAIProvider, GeminiConfig

config = GeminiConfig(api_key="your-api-key")
provider = GeminiAIProvider(config)

# Use with AIChannel
ai_channel = AIChannel("ai", provider=provider)

Install with: pip install roomkit[gemini]

vLLM Provider (Local LLM)¶

VLLMConfig ¶

Bases: BaseModel

Configuration for a local vLLM server.

vLLM exposes an OpenAI-compatible API, so this config is translated into an OpenAIConfig by :func:create_vllm_provider.

Attributes:

Name	Type	Description
`model`	`str`	Model name loaded by the vLLM server (required).
`base_url`	`str`	Base URL of the vLLM OpenAI-compatible endpoint.
`api_key`	`SecretStr`	API key (vLLM usually needs no auth).
`max_tokens`	`int`	Maximum tokens in the response.
`temperature`	`float`	Sampling temperature.
`timeout`	`float`	HTTP request timeout in seconds. Increase for servers that load models on first request (e.g. Ollama cold start).

include_stream_usage `class-attribute` `instance-attribute` ¶

include_stream_usage = False

When True, request token usage in streaming responses.

create_vllm_provider ¶

create_vllm_provider(config)

Create an OpenAI-compatible AI provider pointed at a local vLLM server.

This is a factory function — no new subclass is needed because vLLM implements the OpenAI Chat Completions API. The openai SDK is imported lazily when :class:OpenAIAIProvider is instantiated.

Parameters:

Name	Type	Description	Default
`config`	`VLLMConfig`	vLLM connection settings.	required

Returns:

Name	Type	Description
`An`	`OpenAIAIProvider`	class:`OpenAIAIProvider` configured for the local vLLM server.

Usage¶

from roomkit import create_vllm_provider, VLLMConfig
from roomkit.channels.ai import AIChannel

# Configure connection to a local vLLM server
config = VLLMConfig(
    model="meta-llama/Llama-3.1-8B-Instruct",
    base_url="http://localhost:8000/v1",
)

# Factory returns an OpenAIAIProvider pointed at your vLLM server
provider = create_vllm_provider(config)

# Use with AIChannel like any other AI provider
ai_channel = AIChannel("ai", provider=provider)

Install with: pip install roomkit[vllm]

Anthropic Provider¶

AnthropicAIProvider ¶

AnthropicAIProvider(config)

Bases: AIProvider

AI provider using the Anthropic Messages API.

supports_vision `property` ¶

supports_vision

Claude 3+ models support vision.

generate_structured_stream `async` ¶

generate_structured_stream(context)

Yield structured events from the Anthropic Messages streaming API.

When extended thinking is enabled, yields StreamThinkingDelta events before text deltas.

generate `async` ¶

generate(context)

Generate by consuming the structured stream.

generate_stream `async` ¶

generate_stream(context)

Yield text deltas as they arrive from the Anthropic Messages API.

close `async` ¶

close()

Close the underlying HTTP client.

AnthropicConfig ¶

Bases: BaseModel

Anthropic AI provider configuration.

Usage¶

from roomkit import AnthropicAIProvider, AnthropicConfig
from roomkit.channels.ai import AIChannel

config = AnthropicConfig(api_key="your-api-key")
provider = AnthropicAIProvider(config)

ai_channel = AIChannel("ai", provider=provider)

Install with: pip install roomkit[anthropic]

OpenAI Provider¶

OpenAIAIProvider ¶

OpenAIAIProvider(config)

Bases: AIProvider

AI provider using the OpenAI Chat Completions API.

supports_vision `property` ¶

supports_vision

GPT-4o and GPT-4-turbo models support vision.

generate_structured_stream `async` ¶

generate_structured_stream(context)

Yield structured events with <think> tag parsing.

Text inside <think>...</think> is yielded as :class:StreamThinkingDelta; everything else as :class:StreamTextDelta. Tool calls are collected from the final chunks and yielded as :class:StreamToolCall.

generate_stream `async` ¶

generate_stream(context)

Yield text deltas (thinking content filtered out).

close `async` ¶

close()

Close the underlying HTTP client.

OpenAIConfig ¶

Bases: BaseModel

OpenAI AI provider configuration.

Attributes:

Name	Type	Description
`api_key`	`SecretStr`	API key for authentication.
`base_url`	`str \| None`	Custom base URL for OpenAI-compatible APIs (e.g., Ollama, LM Studio, Azure OpenAI, or other providers). If None, uses the default OpenAI API.
`model`	`str`	Model identifier to use.
`max_tokens`	`int`	Maximum tokens in the response.
`temperature`	`float`	Sampling temperature.

timeout `class-attribute` `instance-attribute` ¶

timeout = 120.0

HTTP request timeout in seconds. Increase for local servers that load models on first request (e.g. Ollama).

include_stream_usage `class-attribute` `instance-attribute` ¶

include_stream_usage = False

When True, request token usage in streaming responses via stream_options.include_usage. The usage is included in the final :class:StreamDone event.

Usage¶

from roomkit import OpenAIAIProvider, OpenAIConfig
from roomkit.channels.ai import AIChannel

config = OpenAIConfig(api_key="your-api-key")
provider = OpenAIAIProvider(config)

ai_channel = AIChannel("ai", provider=provider)

Install with: pip install roomkit[openai]

Mistral Provider¶

MistralAIProvider ¶

MistralAIProvider(config)

Bases: AIProvider

AI provider using the Mistral AI API.

Supports streaming, tool calling, vision (Pixtral models), and <think> tag parsing for reasoning models.

supports_vision `property` ¶

supports_vision

Pixtral models support vision.

generate_structured_stream `async` ¶

generate_structured_stream(context)

Yield structured events with <think> tag parsing.

Text inside <think>...</think> is yielded as :class:StreamThinkingDelta; everything else as :class:StreamTextDelta. Tool calls are accumulated from deltas and yielded as :class:StreamToolCall.

generate `async` ¶

generate(context)

Generate by consuming the structured stream.

generate_stream `async` ¶

generate_stream(context)

Yield text deltas as they arrive from the Mistral API.

close `async` ¶

close()

Close the underlying HTTP client.

MistralConfig ¶

Bases: BaseModel

Mistral AI provider configuration.

Attributes:

Name	Type	Description
`api_key`	`SecretStr`	Mistral API key for authentication.
`model`	`str`	Model identifier (e.g. `'mistral-large-latest'`, `'pixtral-large-latest'`).
`max_tokens`	`int`	Maximum tokens in the response.
`temperature`	`float`	Sampling temperature.
`server_url`	`str \| None`	Custom base URL for Mistral-compatible APIs. If `None`, uses the default Mistral endpoint.

Usage¶

from roomkit import MistralAIProvider, MistralConfig
from roomkit.channels.ai import AIChannel

config = MistralConfig(api_key="your-api-key")
provider = MistralAIProvider(config)

ai_channel = AIChannel("ai", provider=provider)

Install with: pip install roomkit[mistral]

Azure Provider¶

AzureAIProvider ¶

AzureAIProvider(config)

Bases: OpenAIAIProvider

AI provider using Azure AI Studio's OpenAI-compatible Chat Completions API.

Subclasses :class:OpenAIAIProvider — only client initialisation and provider name differ. All message building, tool handling, response parsing, and streaming are inherited.

AzureAIConfig ¶

Bases: BaseModel

Azure AI Studio provider configuration.

Uses the OpenAI-compatible Chat Completions API exposed by Azure AI Foundry deployments (DeepSeek, GPT-4o, Mistral, etc.).

Attributes:

Name	Type	Description
`api_key`	`SecretStr`	Azure API key for authentication.
`azure_endpoint`	`str`	Azure AI Foundry project endpoint URL.
`api_version`	`str`	Azure API version string.
`model`	`str`	Deployment name (no default — user must specify).
`max_tokens`	`int`	Maximum tokens in the response.
`temperature`	`float`	Sampling temperature.
`timeout`	`float`	HTTP request timeout in seconds.

include_stream_usage `class-attribute` `instance-attribute` ¶

include_stream_usage = False

When True, request token usage in streaming responses.

Usage¶

from roomkit import AzureAIProvider, AzureAIConfig
from roomkit.channels.ai import AIChannel

config = AzureAIConfig(
    azure_endpoint="https://your-resource.openai.azure.com/",
    api_key="your-api-key",
    deployment="your-deployment-name",
)
provider = AzureAIProvider(config)

ai_channel = AIChannel("ai", provider=provider)

Install with: pip install roomkit[azure]

Streaming¶

StreamEvent `module-attribute` ¶

StreamEvent = StreamThinkingDelta | StreamTextDelta | StreamToolCall | StreamDone

StreamTextDelta ¶

Bases: BaseModel

A text delta from a streaming AI response.

StreamThinkingDelta ¶

Bases: BaseModel

A thinking/reasoning delta from a streaming AI response.

Emitted before text deltas when the model is reasoning.

StreamToolCall ¶

Bases: BaseModel

A complete tool call extracted from a streaming AI response.

StreamDone ¶

Bases: BaseModel

Signals the end of a streaming AI response.

Response Parts¶

AIThinkingPart ¶

Bases: BaseModel

AI reasoning/thinking block in conversation history.

Used to preserve thinking blocks across tool-loop turns (required by providers like Anthropic that mandate round-trip fidelity).

Attributes:

Name	Type	Description
`thinking`	`str`	The reasoning text produced by the model.
`signature`	`str \| None`	Provider-specific opaque token for caching/validation (e.g. Anthropic's thinking block signature).

AIToolCallPart ¶

Bases: BaseModel

Assistant's tool call in conversation history.

AIToolResultPart ¶

Bases: BaseModel

Tool execution result in conversation history.

ProviderError ¶

ProviderError(message, *, retryable=False, provider='', status_code=None)

Bases: Exception

Error from an AI provider SDK call.

Attributes:

Name	Type	Description
`retryable`		Whether the caller should retry the request.
`provider`		Name of the provider that raised the error.
`status_code`		HTTP status code from the provider, if available.

AI Providers¶

AIProvider ¶

name property ¶

supports_vision property ¶

supports_streaming property ¶

supports_structured_streaming property ¶

model_name abstractmethod property ¶

generate abstractmethod async ¶

generate_stream async ¶

generate_structured_stream async ¶

close async ¶

AIContext ¶

AIMessage ¶

AITextPart ¶

AIImagePart ¶

AITool ¶

AIToolCall ¶

AIResponse ¶

MockAIProvider ¶

generate_stream async ¶

generate_structured_stream async ¶

Per-Room AI Configuration¶

Silent Observer Pattern (Meeting Notes)¶

Tools/Function Calling¶

Gemini Provider¶

GeminiAIProvider ¶

supports_vision property ¶

generate_structured_stream async ¶

generate async ¶

generate_stream async ¶

close async ¶

GeminiConfig ¶

Usage¶

vLLM Provider (Local LLM)¶

VLLMConfig ¶

include_stream_usage class-attribute instance-attribute ¶

create_vllm_provider ¶

Usage¶

Anthropic Provider¶

AnthropicAIProvider ¶

supports_vision property ¶

generate_structured_stream async ¶

generate async ¶

generate_stream async ¶

close async ¶

AnthropicConfig ¶

Usage¶

OpenAI Provider¶

OpenAIAIProvider ¶

supports_vision property ¶

generate_structured_stream async ¶

generate_stream async ¶

close async ¶

OpenAIConfig ¶

timeout class-attribute instance-attribute ¶

include_stream_usage class-attribute instance-attribute ¶

Usage¶

Mistral Provider¶

MistralAIProvider ¶

supports_vision property ¶

generate_structured_stream async ¶

generate async ¶

generate_stream async ¶

close async ¶

MistralConfig ¶

Usage¶

Azure Provider¶

AzureAIProvider ¶

AzureAIConfig ¶

include_stream_usage class-attribute instance-attribute ¶

Usage¶

Streaming¶

StreamEvent module-attribute ¶

StreamTextDelta ¶

StreamThinkingDelta ¶

StreamToolCall ¶

StreamDone ¶

Response Parts¶

AIThinkingPart ¶

AIToolCallPart ¶

name `property` ¶

supports_vision `property` ¶

supports_streaming `property` ¶

supports_structured_streaming `property` ¶

model_name `abstractmethod` `property` ¶

generate `abstractmethod` `async` ¶

generate_stream `async` ¶

generate_structured_stream `async` ¶

close `async` ¶

generate_stream `async` ¶

generate_structured_stream `async` ¶

supports_vision `property` ¶

generate_structured_stream `async` ¶

generate `async` ¶

generate_stream `async` ¶

close `async` ¶

include_stream_usage `class-attribute` `instance-attribute` ¶

supports_vision `property` ¶

generate_structured_stream `async` ¶

generate `async` ¶

generate_stream `async` ¶

close `async` ¶

supports_vision `property` ¶

generate_structured_stream `async` ¶

generate_stream `async` ¶

close `async` ¶

timeout `class-attribute` `instance-attribute` ¶

include_stream_usage `class-attribute` `instance-attribute` ¶

supports_vision `property` ¶

generate_structured_stream `async` ¶

generate `async` ¶

generate_stream `async` ¶

close `async` ¶

include_stream_usage `class-attribute` `instance-attribute` ¶

StreamEvent `module-attribute` ¶