Embedding the Runtime

While most users access MUXI through the server, you can embed the runtime directly into your Python application. This enables custom deployments, integration with existing systems, and streamlined local development.

When to embed vs. use the server:

Example: You have an existing Django app and want to add an AI assistant. Instead of running a separate MUXI server and making HTTP calls, embed the runtime directly - your assistant runs in-process with zero network overhead.

Installation

pip install muxi-runtime

Basic Usage

from muxi.runtime import Formation, Config

# Load formation
formation = Formation.load("./my-formation")

# Send a message
response = formation.chat("Hello!")
print(response.text)

Configuration

from muxi.runtime import Formation, Config

config = Config(
    port=8001,
    host="0.0.0.0",
    log_level="info"
)

formation = Formation.load(
    "./my-formation",
    config=config
)

Running as Server

from muxi.runtime import Formation

formation = Formation.load("./my-formation")
formation.serve(port=8001)

Or with uvicorn:

from muxi.runtime import Formation

formation = Formation.load("./my-formation")
app = formation.as_asgi()

# Run with uvicorn
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8001)

Operational tips

• Load once, reuse: Initialize Formation.load(...) at process start and reuse; avoid per-request loads.
• Secrets via env: Keep secrets in env or secrets.enc; don’t bake tokens in code.
• Graceful shutdown: When embedding in ASGI/WSGI, ensure the process awaits inflight tasks before exit.
• Keep logging wired: Forward runtime logs/events to your app’s logging/observability pipeline for auditability.

Programmatic Access

Chat

response = formation.chat(
    message="Hello!",
    user_id="user_123",
    session_id="sess_abc"
)

print(response.text)
print(response.agent)

Streaming

for chunk in formation.chat_stream("Tell me a story"):
    print(chunk.text, end="")

Sessions

# Create session
session = formation.create_session(user_id="user_123")

# Chat with session
response = formation.chat(
    "Hello",
    session_id=session.id
)

# Get history
history = formation.get_session(session.id)

Agents

# List agents
agents = formation.list_agents()

# Target specific agent
response = formation.chat(
    "Research this",
    agent="researcher"
)

Triggers

response = formation.trigger(
    "github-issue",
    data={
        "repository": "muxi/runtime",
        "issue": {"number": 123}
    }
)

Events

Subscribe to runtime events:

from muxi.runtime import Formation, EventType

formation = Formation.load("./my-formation")

@formation.on(EventType.CHAT_STARTED)
def on_chat(event):
    print(f"Chat started: {event.session_id}")

@formation.on(EventType.CHAT_COMPLETED)
def on_complete(event):
    print(f"Chat completed: {event.response.text[:50]}...")

Custom Integrations

FastAPI

from fastapi import FastAPI
from muxi.runtime import Formation

app = FastAPI()
formation = Formation.load("./my-formation")

@app.post("/chat")
async def chat(message: str):
    response = await formation.chat_async(message)
    return {"response": response.text}

Django

from django.http import JsonResponse
from muxi.runtime import Formation

formation = Formation.load("./my-formation")

def chat_view(request):
    message = request.POST.get("message")
    response = formation.chat(message)
    return JsonResponse({"response": response.text})

Use Cases

1. Microservice - Run as dedicated service
2. Embedded - Within existing application
3. Serverless - AWS Lambda, Cloud Functions
4. Testing - Unit/integration tests
5. Development - Custom dev workflows

Limitations

When embedding:

• No server management features
• No multi-formation support
• Manual scaling
• No built-in monitoring

For production, consider using the full MUXI Server.