Python SDK

uv add mix-python-sdk

import asyncio
import os
from dotenv import load_dotenv
from mix_python_sdk import Mix
from mix_python_sdk.helpers import StreamingSession

async def main():
    load_dotenv()

    async with Mix(server_url=os.getenv("MIX_SERVER_URL")) as mix:
        # Store API key
        mix.authentication.store_api_key(
            api_key=os.getenv("OPENROUTER_API_KEY"),
            provider="openrouter"
        )

        # Auto-managed session with streaming
        async with StreamingSession(mix, title="Demo") as session:
            await session.send(
                "What's the capital of France?",
                on_content=lambda text: print(text, end="", flush=True),
                on_complete=lambda: print("\n")
            )

if __name__ == "__main__":
    asyncio.run(main())

from mix_python_sdk import Mix
import os

with Mix(server_url=os.getenv("MIX_SERVER_URL")) as mix:
    mix.authentication.store_api_key(
        api_key=os.getenv("OPENROUTER_API_KEY"),
        provider="openrouter"
    )

    session = mix.sessions.create(title="Demo")
    response = mix.messages.send(id=session.id, text="Hello")
    print(response.assistant_response)

async def query(
    mix: Mix,
    session_id: str,
    message: str,
) -> AsyncIterator[StreamEvent]

async for event in query(mix, session.id, "What's 2+2?"):
    if event.type == "content":
        print(event.content, end="", flush=True)
    elif event.type == "complete":
        print("\nDone!")

async def send_with_callbacks(
    mix: Mix,
    session_id: str,
    message: str,
    *,
    on_thinking: Optional[Callable[[str], None]] = None,
    on_content: Optional[Callable[[str], None]] = None,
    on_tool: Optional[Callable[[Any], None]] = None,
    on_tool_execution_start: Optional[Callable[[Any], None]] = None,
    on_tool_execution_complete: Optional[Callable[[Any], None]] = None,
    on_error: Optional[Callable[[str], None]] = None,
    on_permission: Optional[Callable[[Any], None]] = None,
    on_complete: Optional[Callable[[], None]] = None,
) -> None

await send_with_callbacks(
    mix,
    session_id=session.id,
    message="Hello!",
    on_content=lambda text: print(text, end="", flush=True),
    on_tool=lambda tool: print(f"\nTool: {tool.name}"),
    on_complete=lambda: print("\nDone!")
)

Mix(
    server_idx: Optional[int] = None,
    server_url: Optional[str] = None,
    url_params: Optional[Dict[str, str]] = None,
    client: Optional[HttpClient] = None,
    async_client: Optional[AsyncHttpClient] = None,
    retry_config: OptionalNullable[RetryConfig] = None,
    timeout_ms: Optional[int] = None,
    debug_logger: Optional[Logger] = None
)

# Synchronous
with Mix(server_url="http://localhost:8088") as mix:
    # Operations here
    pass

# Asynchronous
async with Mix(server_url="http://localhost:8088") as mix:
    # Operations here
    pass

StreamingSession(
    mix: Mix,
    title: str,
    custom_system_prompt: Optional[str] = None
)

# Query pattern (iterator)
async for event in session.query(message="Hello"):
    if event.type == "content":
        print(event.content)

# Callback pattern
await session.send(
    message="Hello",
    on_content=lambda text: print(text),
    on_complete=lambda: print("Done")
)

async with StreamingSession(mix, title="Demo") as session:
    print(f"Session ID: {session.id}")
    await session.send("Hello!", on_content=lambda t: print(t))

class Provider(str, Enum):
    ANTHROPIC = "anthropic"
    OPENAI = "openai"
    OPENROUTER = "openrouter"
    GEMINI = "gemini"
    BRAVE = "brave"

class PromptMode(str, Enum):
    DEFAULT = "default"    # Use base system prompt only
    APPEND = "append"      # Append customSystemPrompt to base (50KB limit)
    REPLACE = "replace"    # Replace base prompt with custom (100KB limit)

class RetryConfig(BaseModel):
    strategy: str                    # Retry strategy ("backoff")
    backoff: BackoffStrategy         # Backoff configuration
    retry_connection_errors: bool    # Whether to retry connection errors

class SessionData(BaseModel):
    id: str                              # Unique session identifier
    title: str                           # Session title
    created_at: date                     # Creation timestamp
    assistant_message_count: int         # Number of assistant messages
    user_message_count: int              # Number of user messages
    tool_call_count: int                 # Number of tool calls
    prompt_tokens: int                   # Total prompt tokens
    completion_tokens: int               # Total completion tokens
    cost: float                          # Total cost
    first_user_message: Optional[str]    # First message (optional)

class BackendMessage(BaseModel):
    id: str                              # Unique message ID
    session_id: str                      # Session ID
    role: str                            # Role: user/assistant/tool
    user_input: str                      # User's input
    assistant_response: Optional[str]    # Assistant's response
    reasoning: Optional[str]             # Reasoning process
    reasoning_duration: Optional[int]    # Reasoning duration (ms)
    tool_calls: List[ToolCallData]       # Tool calls made

class ToolCallData(BaseModel):
    id: str                      # Unique tool call ID
    name: str                    # Tool name
    input: str                   # Input parameters
    result: Optional[str]        # Execution result
    finished: bool               # Whether complete
    is_error: Optional[bool]     # Whether errored
    type: str                    # Tool type

class FileInfo(BaseModel):
    name: str       # File name
    size: int       # Size in bytes
    modified: int   # Unix timestamp
    is_dir: bool    # Is directory
    url: str        # Static access URL

{
    "event": EventTypeEnum,      # Event type identifier
    "id": str,                   # Sequential event ID
    "retry": Optional[int],      # Client retry interval (ms)
    "data": EventData            # Event-specific data
}

class MixError(Exception):
    message: str                  # Error message
    status_code: int              # HTTP status code
    headers: httpx.Headers        # Response headers
    body: str                     # Response body
    raw_response: httpx.Response  # Raw HTTP response
    data: Any                     # Optional structured error data

class ErrorResponse(BaseModel):
    error: RESTError
        code: int        # HTTP status code
        message: str     # Error message
        type: Type       # Error type

class BashResponseMetadata(BaseModel):
    start_time: int  # Unix timestamp (ms) when command started
    end_time: int    # Unix timestamp (ms) when command completed

class ReadTextResponseMetadata(BaseModel):
    file_path: str  # The path/URL that was read
    content: str    # The raw content without line numbers

class GlobResponseMetadata(BaseModel):
    number_of_files: int  # Count of files returned
    truncated: bool       # Whether results were limited to 100 files

class MediaAnalysisType(str, Enum):
    IMAGE = "image"
    AUDIO = "audio"
    VIDEO = "video"
    PDF = "pdf"

class ReadMediaResponse(BaseModel):
    results: List[ReadMediaResult]  # List of analysis results
    summary: str                    # Overall summary

class ReadMediaResult(BaseModel):
    file_path: str            # Path to analyzed file
    media_type: str           # Type of media analyzed
    analysis: str             # Analysis results from AI
    error: Optional[str]      # Error message if failed

class GrepResponseMetadata(BaseModel):
    number_of_matches: int  # Total number of matches found
    truncated: bool         # True if results limited to 100 matches

class WriteResponseMetadata(BaseModel):
    diff: str        # Diff text showing changes
    additions: int   # Number of lines added
    removals: int    # Number of lines removed

class EditResponseMetadata(BaseModel):
    diff: str        # Diff text showing changes
    additions: int   # Number of lines added
    removals: int    # Number of lines removed

class PythonExecutionResult(BaseModel):
    type: str          # Always "code_execution_result"
    stdout: str        # Standard output
    stderr: str        # Standard error
    return_code: int   # Exit code (0 = success)

class SearchType(str, Enum):
    WEB = "web"
    IMAGES = "images"
    VIDEOS = "videos"

class Safesearch(str, Enum):
    STRICT = "strict"
    MODERATE = "moderate"
    OFF = "off"

class BraveSearchResponse(BaseModel):
    type: str
    web: WebResults
        results: List[SearchResult]
            title: str
            url: str
            description: str

class ImageSearchResponse(BaseModel):
    type: str
    results: List[ImageResult]
        title: str
        url: str
        source: str
        thumbnail: ImageResultThumbnail
            src: str
        properties: ImageResultProperties
            url: str

class VideoSearchResponse(BaseModel):
    type: str
    results: List[VideoResult]
        title: str
        url: str
        source: str
        thumbnail: VideoResultThumbnail
            src: str
        properties: VideoResultProperties
            url: str
            duration: str
            views: str

class Todo(BaseModel):
    id: str                # Unique identifier
    content: str           # Task description
    status: TodoStatus     # Current status (pending, in_progress, completed)
    priority: TodoPriority # Priority level (low, medium, high)

class MediaOutput(BaseModel):
    path: Optional[str]           # Absolute path or URL to media file
    type: MediaType               # Type of media (image, video, audio, gsap_animation, pdf, csv)
    title: str                    # Title or name for the output
    description: Optional[str]    # Optional description
    config: Optional[Dict]        # Configuration data (gsap_animation only)
    start_time: Optional[int]     # Start time in seconds (video/audio)
    duration: Optional[int]       # Duration in seconds (video/audio)

async with StreamingSession(mix, title="Multi-turn") as session:
    # First question
    await session.send("What's the capital of France?")

    # Follow-up maintains context
    await session.send("What about Italy?")

    # Context is preserved throughout the session
    await session.send("Which has a larger population?")

# Original session
session = mix.sessions.create(title="Original")
mix.messages.send(id=session.id, text="Hello")
mix.messages.send(id=session.id, text="Tell me about Python")

# Fork at message index 1
forked = mix.sessions.fork(
    id=session.id,
    message_index=1,
    title="Forked - Alternative Path"
)

# Forked session continues from message 1
mix.messages.send(id=forked.id, text="Tell me about Rust instead")

# Rewind to specific message
mix.sessions.rewind_session(
    id=session.id,
    message_id="msg_123",
    cleanup_media=True  # Also delete associated media files
)

# Upload image
with open("photo.jpg", "rb") as f:
    mix.files.upload_session_file(
        id=session_id,
        file={
            "file_name": "photo.jpg",
            "content": f,
            "content_type": "image/jpeg"
        }
    )

# Generate thumbnails
thumb_100 = mix.files.get_session_file(
    id=session_id,
    filename="photo.jpg",
    thumb="100"  # 100x100 box
)

thumb_w200 = mix.files.get_session_file(
    id=session_id,
    filename="photo.jpg",
    thumb="w200"  # Width-constrained to 200px
)

# Main agent: complex reasoning tasks
mix.preferences.update_preferences(
    main_agent_model="openrouter.deepseek-v3.1",
    main_agent_max_tokens=8000,
    main_agent_reasoning_effort="high"
)

# Sub agent: quick supporting tasks
mix.preferences.update_preferences(
    sub_agent_model="openrouter.zai-glm-4.5-air",
    sub_agent_max_tokens=2000,
    sub_agent_reasoning_effort="low"
)

from mix_python_sdk.utils import BackoffStrategy, RetryConfig

retry = RetryConfig(
    "backoff",
    BackoffStrategy(
        initial_interval=1,
        max_interval=50,
        exponent=1.1,
        max_elapsed_time=100
    ),
    retry_connection_errors=False
)

mix = Mix(server_url="http://localhost:8088", retry_config=retry)

import logging

logging.basicConfig(level=logging.DEBUG)
mix = Mix(
    server_url="http://localhost:8088",
    debug_logger=logging.getLogger("mix_python_sdk")
)

# Or use environment variable
# MIX_DEBUG=true

import threading
import time

def stream_message(mix, session_id: str, message: str):
    # Read path: SSE connection
    stream_response = mix.streaming.stream_events(session_id=session_id)
    time.sleep(0.5)  # Allow SSE connection to establish

    # Write path: REST API (separate thread)
    send_thread = threading.Thread(
        target=lambda: mix.messages.send(id=session_id, text=message),
        daemon=True
    )
    send_thread.start()

    # Process events
    with stream_response.result as event_stream:
        for event in event_stream:
            if isinstance(event, SSEContentEvent):
                print(event.data.content, end="", flush=True)
            elif isinstance(event, SSECompleteEvent):
                break

    send_thread.join(timeout=1.0)

from mix_python_sdk import Mix
import os

with Mix(server_url=os.getenv("MIX_SERVER_URL")) as mix:
    # Health check
    health = mix.system.get_health()
    print(f"Status: {health.status}")

    # Store API key
    mix.authentication.store_api_key(
        api_key=os.getenv("OPENROUTER_API_KEY"),
        provider="openrouter"
    )

    # Create and use session
    session = mix.sessions.create(title="Demo")
    response = mix.messages.send(id=session.id, text="Hello")
    print(response.assistant_response)

    # Clean up
    mix.sessions.delete(id=session.id)

import asyncio
from mix_python_sdk import Mix
from mix_python_sdk.helpers import query

async def main():
    async with Mix(server_url="http://localhost:8088") as mix:
        mix.authentication.store_api_key(api_key=api_key, provider="openrouter")
        session = mix.sessions.create(title="Demo")

        async for event in query(mix, session.id, "What's 2+2?"):
            if event.type == "content":
                print(event.content, end="", flush=True)
            elif event.type == "tool":
                print(f"\nTool: {event.tool_name}")

        mix.sessions.delete(id=session.id)

asyncio.run(main())

import asyncio
from mix_python_sdk import Mix
from mix_python_sdk.helpers import send_with_callbacks

async def main():
    async with Mix(server_url="http://localhost:8088") as mix:
        mix.authentication.store_api_key(api_key=api_key, provider="openrouter")
        session = mix.sessions.create(title="Demo")

        await send_with_callbacks(
            mix,
            session_id=session.id,
            message="What's your working directory?",
            on_thinking=lambda text: print(f"Thinking: {text}", end="", flush=True),
            on_content=lambda text: print(f"\nResponse: {text}", end="", flush=True),
            on_tool=lambda tool: print(f"\nTool: {tool.name}"),
            on_tool_execution_complete=lambda data: print(f"Result: {data.progress}"),
            on_error=lambda error: print(f"\nError: {error}"),
            on_complete=lambda: print("\nComplete!")
        )

        mix.sessions.delete(id=session.id)

asyncio.run(main())

# List sessions with metadata
sessions = mix.sessions.list()
for s in sessions:
    print(f"Title: {s.title}")
    print(f"Messages: {s.user_message_count} user, {s.assistant_message_count} assistant")
    print(f"Tokens: {s.prompt_tokens} prompt, {s.completion_tokens} completion")
    print(f"Cost: ${s.cost:.6f}")
    print(f"Tool calls: {s.tool_call_count}")

# Export complete session transcript
export = mix.sessions.export_session(id=session_id)

# Access full message history with tool calls
for msg in export.messages:
    print(f"Role: {msg.role}")
    print(f"Content: {msg.user_input or msg.assistant_response}")
    if msg.reasoning:
        print(f"Reasoning time: {msg.reasoning_duration}ms")
    for tool_call in msg.tool_calls:
        print(f"Tool: {tool_call.name}")
        print(f"Input: {tool_call.input}")
        print(f"Result: {tool_call.result}")

# Get tools status
tools_status = mix.tools.get_tools_status()

for category_name, category_info in tools_status.categories.items():
    print(f"Category: {category_info.display_name}")
    for tool in category_info.tools:
        auth_status = "Ready" if tool.authenticated else "Needs auth"
        print(f"  - {tool.display_name}: {auth_status}")