Connect a LiveKit agent

Start with project credentials. Egma creates a room, dispatches your named worker, runs the conversation, and deletes the room when the simulation ends. The token endpoint option below lets you keep the LiveKit key pair on your own server. You need a working LiveKit agent, access to its LiveKit project, and an initialized Egma project. Complete CLI sign-in first.

1. Prepare the worker

Use the worker and LiveKit project you already run. Set an explicit dispatch name in its startup configuration: agent_name in Python or agentName in JavaScript. The name must exactly match the --livekit-agent-name value you give Egma. A named dispatch cannot reach an unnamed worker. For example, use front-desk as the name in both places. Keep the worker running while the tests execute. A local worker can join simulations if it connects to the configured LiveKit project. See LiveKit’s agent dispatch guide. Install the SDK for your language and add await simulation(agent, ctx, session) after you create the agent and session, before session.start. This is required for every LiveKit simulation, including tests without mock tools. It reports your agent’s tools, applies the test’s mocks, and sends the agent’s conversation evidence to Egma. It does nothing in production rooms.

LiveKit Python SDK

Install the package and add the simulation hook to your Python worker.

LiveKit JavaScript SDK

Install the package and add the simulation hook to your JavaScript or TypeScript worker.

Create a project API key from your initialized agent repository:

egma project api-key create --name "LiveKit worker"

Copy the key when the CLI prints it; it is shown only once. Add these settings to the worker through your secret store:

Variable	Value
`EGMA_URL`	`https://app.egma.ai`, or a URL for your self-hosted Egma instance that the worker can reach.
`EGMA_API_KEY`	The project API key you just created.

These settings send the agent’s evidence to the same Egma project that runs the test. They are separate from the LiveKit project credentials used below. Restart or redeploy the worker after adding the hook and settings.

2. Register the agent in Egma

Check egma/config.yaml. Reuse the existing Egma agent if it is already registered. Otherwise, run:

egma agent register --platform livekit --name "Front desk"

Set EGMA_AGENT_ID to the Egma agent ID printed by the command. This ID is separate from your LiveKit worker’s dispatch name.

3. Add a voice connection

Load your existing LiveKit project settings from your secret store. In the commands below, LIVEKIT_URL is the project’s server URL and LIVEKIT_AGENT_NAME is the explicit worker name from step 1.

export EGMA_LIVEKIT_API_KEY="$LIVEKIT_API_KEY"
export EGMA_LIVEKIT_API_SECRET="$LIVEKIT_API_SECRET"

egma agent connection add \
  --agent "$EGMA_AGENT_ID" \
  --access livekit-project-credentials \
  --modality voice \
  --livekit-url "$LIVEKIT_URL" \
  --livekit-agent-name "$LIVEKIT_AGENT_NAME" \
  --name "LiveKit voice"

The key must belong to the project that runs your worker and permit room and agent-dispatch operations. Egma stores the credentials separately from egma/config.yaml. For credentials supplied by another process, add --credentials-stdin and send a JSON object with apiKey and apiSecret fields through standard input. Do not put the secret in command-line arguments. Keep the connection ID as EGMA_CONNECTION_ID, then follow the CLI guide to write a test and start a run. A voice simulation waits for your worker to join and publish audio.

Add a text connection

Text mode tests conversation logic without the speech stack. First configure your worker’s text input and transcription output using the chat example in the Python SDK or JavaScript SDK guide. In rooms whose names start with egma-sim-chat-, disable audio input and output. Keep any separate greeting or audio publisher off in that mode. Then add a second connection:

egma agent connection add \
  --agent "$EGMA_AGENT_ID" \
  --access livekit-project-credentials \
  --modality chat \
  --livekit-url "$LIVEKIT_URL" \
  --livekit-agent-name "$LIVEKIT_AGENT_NAME" \
  --name "LiveKit text"

Use this connection’s ID when starting a text run. The same suite can run against either connection.

Pass data to a test

If your worker reads startup context from ctx.job.metadata, put that context in the test’s ## Env section:

## Env

```json
{
  "job_dispatch_metadata": {
    "tenant": "oak-street",
    "locale": "en-US"
  }
}
```

Use the exact keys your worker reads. Egma serializes this object as the dispatch metadata string. Your normal JSON.parse or json.loads call reads it in the worker. A test without this field receives empty dispatch metadata. Egma does not add scenario instructions or other Egma fields to it. To detect a simulation, check whether the room name starts with egma-sim-. This includes text rooms. Keep that prefix reserved for Egma when your own application creates production rooms.

Use a token endpoint

Choose this option when your server must mint the participant token instead of giving Egma a LiveKit project key pair. The endpoint must return access to the requested room and arrange the named worker’s dispatch. The example below does both through LiveKit’s room configuration. For this authentication path, your endpoint owns the room lifecycle. It prepares the room, dispatches the requested worker, and cleans the room up after Egma leaves.

Create the endpoint

This example uses Python and FastAPI. Install its dependencies in a server project:

uv add livekit-api fastapi uvicorn

Set these environment variables on that server:

Variable	Value
`LIVEKIT_URL`	Your public LiveKit server URL.
`LIVEKIT_API_KEY`	The LiveKit project API key.
`LIVEKIT_API_SECRET`	The matching project secret.
`LIVEKIT_AGENT_NAME`	The allowed worker’s explicit dispatch name.
`EGMA_TOKEN_SECRET`	A strong secret used only to authenticate Egma to this endpoint.

Create token_server.py:

import os
import secrets
from datetime import timedelta

from fastapi import FastAPI, Header, HTTPException
from google.protobuf.json_format import ParseDict
from livekit import api
from pydantic import BaseModel, Field

app = FastAPI()
endpoint_secret = os.environ["EGMA_TOKEN_SECRET"]
worker_name = os.environ["LIVEKIT_AGENT_NAME"]
server_url = os.environ["LIVEKIT_URL"]


class Dispatch(BaseModel):
    agent_name: str
    metadata: str | None = None


class RoomConfig(BaseModel):
    agents: list[Dispatch] = Field(min_length=1, max_length=1)


class TokenRequest(BaseModel):
    room_name: str = Field(pattern=r"^egma-sim-.+", max_length=255)
    participant_identity: str = Field(pattern=r"^egma-persona-.+", max_length=255)
    participant_name: str = Field(max_length=255)
    room_config: RoomConfig


@app.post("/egma/livekit-token", status_code=201)
async def create_token(body: TokenRequest, authorization: str = Header(default="")):
    expected = f"Bearer {endpoint_secret}".encode()
    if not secrets.compare_digest(authorization.encode(), expected):
        raise HTTPException(status_code=401, detail="Unauthorized")

    if body.room_config.agents[0].agent_name != worker_name:
        raise HTTPException(status_code=400, detail="Unknown agent")

    room_config = body.room_config.model_dump(exclude_none=True)
    room_config["empty_timeout"] = 60
    token = (
        api.AccessToken(
            os.environ["LIVEKIT_API_KEY"], os.environ["LIVEKIT_API_SECRET"]
        )
        .with_identity(body.participant_identity)
        .with_name(body.participant_name)
        .with_grants(api.VideoGrants(
            room_join=True,
            room=body.room_name,
            can_publish=True,
            can_subscribe=True,
            can_publish_data=True,
        ))
        .with_room_config(ParseDict(room_config, api.RoomConfiguration()))
        .with_ttl(timedelta(minutes=5))
    )
    return {"server_url": server_url, "participant_token": token.to_jwt()}

Start the server and expose this route through your HTTPS reverse proxy:

uv run uvicorn token_server:app --host 0.0.0.0 --port 8080

The endpoint preserves the requested worker and its test metadata in the token’s room configuration. Do not pre-create the room: LiveKit applies that configuration when the first participant creates it by joining. This follows LiveKit’s token endpoint contract. The endpoint and returned LiveKit address must resolve to public addresses. Use HTTPS for the endpoint and WSS or HTTPS for LIVEKIT_URL. These rules also apply to self-hosted Egma. For a private-network LiveKit server, use project credentials instead.

Connect Egma to the endpoint

In your agent repository, set LIVEKIT_TOKEN_ENDPOINT to the full public URL ending in /egma/livekit-token. Load the same EGMA_TOKEN_SECRET from your secret store. This command passes the authorization header through standard input:

python3 -c 'import json, os; print(json.dumps({"headers": {"Authorization": "Bearer " + os.environ["EGMA_TOKEN_SECRET"]}}))' |
  egma agent connection add \
    --agent "$EGMA_AGENT_ID" \
    --access livekit-token-endpoint \
    --modality voice \
    --livekit-agent-name "$LIVEKIT_AGENT_NAME" \
    --livekit-token-endpoint "$LIVEKIT_TOKEN_ENDPOINT" \
    --name "LiveKit token endpoint" \
    --credentials-stdin

For a text connection, use --modality chat after adding the worker’s text configuration described above. Use the returned connection ID to start a run with the CLI. Check that the worker joins the egma-sim- room and receives any test metadata. Egma makes one token request per simulation. It allows 20 seconds for the response, does not follow redirects, and accepts a response body up to 64 KiB. It joins the room once with the participant token. A voice simulation publishes and subscribes to audio. A chat simulation joins as one text-only client: it publishes no media tracks, subscribes to no audio, and performs no audio decoding, speech-to-text, text-to-speech, recording, or voice processing. Egma waits 30 seconds for your worker to join and answer, then ends the simulation as agent_never_joined. Egma leaves when the conversation ends. It never deletes the room and never uses the token again, so your token endpoint owns cleanup for this room. Configure your worker to shut down when the caller leaves. A short empty-room timeout on your LiveKit project is one way to clean the room up.

If the worker does not join or respond

Check the worker’s registered name against LIVEKIT_AGENT_NAME and confirm it is connected to the same project as the connection credentials.
Check the worker logs for missing dispatch metadata or startup errors.
Confirm that the worker calls simulation before session.start, has a project API key, and can reach EGMA_URL. A NotReported error means the worker could not complete its startup exchange with Egma; the session does not start.
For voice, confirm that the worker publishes audio. For text, check its text input and transcription output rather than waiting for an audio track.
For a token endpoint, check the HTTP status, public DNS, returned server URL, and room configuration. An issued token alone does not prove dispatch worked.

The simulation transcript uses the agent’s own turns and tool calls. Calls answered by a test mock are marked mocked; other tools run their real implementation and their calls are also recorded. The voice recording keeps what the simulated caller heard. To capture real conversations, add monitor(ctx) as described in Set up monitoring. Both hooks can stay in one worker: simulation acts only in simulation rooms, and monitor acts only in production rooms.

Get started

Integrations

Testing

Monitoring

Grading

Connect a LiveKit agent

1. Prepare the worker

LiveKit Python SDK

LiveKit JavaScript SDK

2. Register the agent in Egma

3. Add a voice connection

Add a text connection

Pass data to a test

Use a token endpoint

Create the endpoint

Connect Egma to the endpoint

If the worker does not join or respond

​1. Prepare the worker

LiveKit Python SDK

LiveKit JavaScript SDK

​2. Register the agent in Egma

​3. Add a voice connection

​Add a text connection

​Pass data to a test

​Use a token endpoint

​Create the endpoint

​Connect Egma to the endpoint

​If the worker does not join or respond

1. Prepare the worker

2. Register the agent in Egma

3. Add a voice connection

Add a text connection

Pass data to a test

Use a token endpoint

Create the endpoint

Connect Egma to the endpoint

If the worker does not join or respond