moss-agent

Moss runtime for LiveKit voice agents - a hot index cache shared across rooms and a one-line attach per call.

One process-wide MossAgent holds auth and the loaded indexes. Every conversation gets its own MossCall, and all of them query the same warm in-memory indexes.

Install

pip install moss-agent
# with LiveKit, for attach(ctx):
pip install "moss-agent[livekit]"

Use - voice (LiveKit)

from livekit.agents import AgentServer, JobContext, JobProcess
from moss_agent import MossAgent

server = AgentServer()


def prewarm(proc: JobProcess) -> None:
    agent = MossAgent(
        project_id="...",
        project_key="...",
    )
    import asyncio
    asyncio.run(agent.load_indexes(["product_catalog", "faq", "policies"]))
    proc.userdata["moss_agent"] = agent


server.setup_fnc = prewarm


@server.rtc_session(agent_name="my-agent")
async def handle_visit(ctx: JobContext) -> None:
    await ctx.connect()

    agent: MossAgent = ctx.proc.userdata["moss_agent"]
    call = agent.attach(ctx)

    results = await call.query("product_catalog", user_question)
    # ... your existing agent loop ...

attach(ctx) uses the room name as the call's room ID, the job ID as its session ID and the room's creation time as its start, and ends the call when LiveKit shuts the room down.

Use - voice (any framework)

agent = MossAgent(project_id="...", project_key="...", cache_path="/var/cache/moss")
await agent.load_indexes(["product_catalog", "faq"])

# When a conversation starts, with its ID from your voice stack:
call = agent.start_call(room_id, session_id=session_id)
results = await call.query("product_catalog", user_question)
results = await call.query_multi_index(["product_catalog", "faq"], user_question)

# When it ends:
await call.end()

start_call is idempotent while the call is active: starting the same room_id again returns a handle to the same call. started_at (a timezone-aware datetime, default now) is the call's start time; a process that joins a conversation already in progress should pass the real start time.

Ambient retrieval (STT to Moss)

Instead of waiting for the LLM to call a search tool, query Moss with every final user transcript and give the results to the LLM with the question. Each turn needs one LLM pass instead of two, and the Moss query takes a few milliseconds, so the user never waits on it.

One helper turns a transcript into a context block:

from moss_agent import MossCall, QueryOptions


async def ground(call: MossCall, transcript: str) -> str:
    """Moss results for one user turn, or "" when nothing is relevant."""
    if not transcript.strip():
        return ""
    try:
        results = await call.query(
            "faq", transcript, QueryOptions(top_k=3, min_score=0.4)
        )
    except Exception:
        return ""  # a failed lookup must never break the voice turn
    if not results.docs:
        return ""
    passages = "\n".join(f"[{i}] {doc.text}" for i, doc in enumerate(results.docs, 1))
    return f"Relevant knowledge from Moss:\n{passages}"

min_score drops weak matches, so small talk such as "yes" or "thanks" adds nothing to the prompt. Tune it on your own queries.

LiveKit: on_user_turn_completed runs after the user's final transcript and before the LLM answers.

from livekit.agents import Agent, AgentSession, ChatContext, ChatMessage, JobContext


class Assistant(Agent):
    def __init__(self, call: MossCall) -> None:
        super().__init__(instructions="Answer using the Moss knowledge you are given.")
        self._call = call

    async def on_user_turn_completed(
        self, turn_ctx: ChatContext, new_message: ChatMessage
    ) -> None:
        context = await ground(self._call, new_message.text_content or "")
        if context:
            turn_ctx.add_message(role="assistant", content=context)


@server.rtc_session(agent_name="my-agent")
async def handle_visit(ctx: JobContext) -> None:
    await ctx.connect()
    call = ctx.proc.userdata["moss_agent"].attach(ctx)
    session = AgentSession(...)  # your STT, LLM and TTS
    await session.start(agent=Assistant(call), room=ctx.room)

Any framework: call ground from your STT's final-transcript callback and send the result to the LLM with the transcript.

async def on_final_transcript(text: str) -> None:
    context = await ground(call, text)
    llm_input = f"{context}\n\n[Current User Question]\n{text}" if context else text
    await send_to_llm(llm_input)  # your framework's LLM input

Use - text (HTTP / chat / non-voice)

from moss_agent import MossAgent

# Module level - reuse across requests
agent = MossAgent(project_id="...", project_key="...")
await agent.load_indexes(["product_catalog"])


# In your HTTP / chat handler
async def handle_message(req):
    results = await agent.query("product_catalog", req.user_message)
    # ... feed results to your model ...

Query options

QueryOptions works the same on MossCall and on the text path:

• top_k (default 5, 10 for multi-index) and alpha (default 0.8 for single and multi-index; 1.0 is embedding-only, 0.0 keyword-only).
• min_score drops results whose score (relevance from 0 to 1) is below it. Keyword-only queries (alpha=0.0) have no embedding to score against, so min_score raises an error on them.
• candidate_depth sets how many hits a hybrid query ranks on each signal before fusion (default 2 * top_k).
• embedding uses your own query vector, and filter applies a metadata filter.

from moss_agent import QueryOptions

results = await call.query("faq", question, QueryOptions(top_k=3, min_score=0.4))

Adding documents

create_index and add_docs take documents as DocumentInfo objects, dicts with the same fields, JSON array or JSON Lines text, or file paths:

from pathlib import Path

await agent.create_index("faqs", '[{"id": "1", "text": "Track your order in your account."}]')
await agent.create_index("faqs-jsonl", Path("faqs.jsonl"))
await agent.add_docs("faqs", [{"id": "2", "text": "30-day returns.", "metadata": {"days": 30}}])
await agent.create_index("handbook", [Path("docs/handbook.md"), Path("docs/faq.txt")])

• .txt, .md and .markdown files are chunked natively. Chunk ids are {path}#{n} and each chunk's metadata has parentId.
• Integer ids and number or boolean metadata values become strings; booleans are stored as "true" and "false".
• chunker= splits every document's text with your own function. Chunks may have at most 2,000 characters.
• chunking=Chunking(chunk_size=800, chunk_overlap=100) tunes the native chunker (chunk_size 200 to 2000, chunk_overlap 0 to half of chunk_size, mode="pages" or "sentences", markdown_header_depth 1 to 6) and also chunks DocumentInfo, dict and JSON input. Pass chunker or chunking, not both; neither works with model_id="custom".

create_index(..., wait=False) submits the build and returns a JobHandle right away:

job = await agent.create_index("faqs", docs, wait=False)
print((await job.status()).progress)
await job.wait()  # or later: await agent.wait_for_job(job.job_id)

PDF and DOCX files go through create_index_from_files, with optional ParseOptions:

from moss_agent import ParseFileInput, ParseOptions

await agent.create_index_from_files(
    "manuals",
    [ParseFileInput(name="manual.pdf", content_type="application/pdf", path="manual.pdf")],
    parse_options=ParseOptions(ocr_mode="auto_ocr", chunk_size=800),
)

Disk cache and offline loads

cache_path on the constructor is the default on-disk index cache for load_index / load_indexes; a per-call cache_path overrides it. Later loads reuse the cached copy while the cloud version is unchanged. With a snapshot on disk, a load that cannot reach the cloud serves the cached snapshot and starts a refresh poller; was_served_stale(name) reports whether an index is served that way.

Web sources

Crawl a site into an index and keep it refreshed. These call the Moss manage API (/v1/manage); set MOSS_CLOUD_API_BASE_URL to use a non-default host (default https://service.usemoss.dev). The constructor's base_url does not apply to web sources, so an agent pointed at another environment needs MOSS_CLOUD_API_BASE_URL set to match.

created = await agent.create_web_source(
    "https://docs.example.com",
    "knowledge-base",
    max_pages=200,
    refresh_cadence="weekly",
)
await agent.wait_for_job(created.job_id)
await agent.load_index("knowledge-base")

sources = await agent.list_web_sources(index_name="knowledge-base")
source = await agent.get_web_source(created.id)
updated = await agent.update_web_source(created.id, refresh_cadence="daily")
resync = await agent.resync_web_source(created.id)
deleted = await agent.delete_web_source(created.id)

A loaded index is a snapshot: after a crawl finishes, call load_index again, or load it once with auto_refresh=True. Failed manage calls raise ManageApiError with the HTTP status. A redirect fails the same way instead of being followed, so the project key is never sent to another URL.

Shutdown

await agent.close() ends every active call, sends the final usage telemetry and releases the agent; afterwards every method raises RuntimeError. MossAgent also works as async with and with.

In a LiveKit worker, attach already ends each call from the room's shutdown callback, so you don't need close() per room. Call close() (or use async with) wherever you own shutdown: a script, a web server's shutdown handler, or tests. If the process exits without it, the core ends open calls when the agent object is freed, which at interpreter exit is best-effort; a killed process sends nothing more.

API surface

• MossAgent(project_id, project_key, base_url=None, client_id=None, *, cache_path=None) - process-wide instance. base_url sets the index service; web sources use MOSS_CLOUD_API_BASE_URL.
• Voice: attach(ctx) -> MossCall (LiveKit) and start_call(room_id, *, session_id=None, started_at=None) -> MossCall. MossCall.query(name, query, options=None), MossCall.query_multi_index(names, query, options=None), MossCall.end(); properties call_id, room_id, session_id, start_at_unix_ms, has_ended (livekit_room_id / livekit_session_id are aliases).
• Text: MossAgent.query(name, query, options=None) and MossAgent.query_multi_index(names, query, options=None).
• Index CRUD: create_index, create_index_from_files, add_docs, delete_docs, delete_index, list_indexes, get_index, get_docs, get_job_status, wait_for_job.
• Cache lifecycle: load_index, load_indexes, unload_index, unload_indexes, was_served_stale.
• Web sources: create_web_source, list_web_sources, get_web_source, update_web_source, resync_web_source, delete_web_source.
• close(), active_call_count().
• Errors: MossError (with stable .code), InvalidArgumentError (also a ValueError), ManageApiError for web sources.

Telemetry

moss-agent always sends usage telemetry to the Moss cloud. See Telemetry for what it contains and when it is sent.

Requirements

• Python 3.10+
• inferedge-moss-core == 0.32.0
• livekit-agents >= 0.10.0 (only required if you call attach(); install via the [livekit] extra)

License

Proprietary. See LICENSE.txt.