Egma Python SDK

Test a LiveKit agent with mock tools and send its production spans to Egma.

The two verbs are separate, and the room decides which one acts:

• simulation(...) is for an Egma simulation room. It reports your agent's tools to Egma, lets Egma answer the tools the running test mocks, and sends the agent's own spans to Egma as that simulation's agent POV. It does nothing in production.
• monitor(...) sends production LiveKit spans to the Monitoring page. It does nothing in an Egma simulation.

Calling one verb never enables or changes the other.

simulation(...) is required for a LiveKit simulation, and it fails closed: an agent that cannot report to Egma raises egma.NotReported and the session does not start. Egma ends that simulation from its own side with the same finding.

Install

pip install 'egma @ git+https://github.com/egma-ai/egma.git#subdirectory=sdks/python'

This source request has no version, tag, or commit. Let the repository's package manager resolve and lock the latest compatible SDK.

For a LiveKit agent on Python 3.11 or newer.

Production monitoring

Open Agents → Connect an agent, choose Monitoring or Both, and then choose LiveKit. The guided flow shows these worker, key, and deployment steps. A LiveKit agent pushes its own spans, so there is nothing to switch on in Egma.

Set the Egma API origin and an existing project API key in the agent's environment:

export EGMA_URL=https://api.egma.ai
export EGMA_API_KEY=egma_sk_...

For self-hosted Egma, prefer the published API address, such as http://localhost:3100. The web address also works because Egma forwards /v1/traces to the API, but the API is one hop shorter. The agent process must be able to make HTTP or HTTPS requests to the address you choose.

In a customer-hosted worker, set these values through your normal deployment secret store. For LiveKit Cloud, place them in a gitignored secrets file and apply it to the agent:

lk agent update-secrets --secrets-file=.env.monitoring

Use the same file with lk agent create --secrets-file=.env.monitoring for a new deployment. LiveKit Cloud restarts the agent after a secret update.

Call monitor as the first statement of the job entrypoint, before ctx.connect and AgentSession.start:

from egma import monitor
from livekit import agents
from livekit.agents import Agent, AgentSession


async def entrypoint(ctx: agents.JobContext) -> None:
    monitor(ctx)
    await ctx.connect()

    agent = Agent(instructions=INSTRUCTIONS, tools=[check_calendar])
    session = AgentSession(stt=..., llm=..., tts=...)
    await session.start(agent=agent, room=ctx.room)

You can pass the same values directly when environment variables are not the right configuration source:

monitor(ctx, endpoint="https://api.egma.ai", api_key=project_key)

Use the same call for an agent hosted in your cloud and an agent hosted in LiveKit Cloud. The helper adds Egma to a compatible OpenTelemetry provider; it does not remove LiveKit Cloud observability or another existing span processor. It sends spans in batches and flushes the final batch when the LiveKit job stops.

If this job runs in an Egma simulation room, the helper returns without adding a production exporter and says so at WARNING. simulation(...) exports that conversation instead, filed under the simulation rather than as a second production call.

The helper reads the room's name for that and nothing else. Every Egma simulation room is named egma-sim-…; a room named anything else gets the production exporter. The name arrives with the job, so this decision costs no network and happens before ctx.connect().

After deployment, open Monitoring. Nothing needs to be confirmed in Egma — the deployed helper is what sends spans, and the first production conversation appears in the list as soon as it arrives.

The caller's phone, SIP, or WebRTC entry path does not change this setup. The SDK does not guess that path from a LiveKit trace.

Setup status

The helper stops immediately with one direct error when required settings are missing or malformed, or when tracer providers conflict. It does not include the key in that error.

After deployment, Monitoring stays empty until the first trace reaches Egma, then lists each production conversation as it arrives. This release uses one-way OTLP export. Egma cannot see a DNS, firewall, or network failure inside the worker. Check the worker's OpenTelemetry logs and confirm that it can reach EGMA_URL when nothing appears.

Simulation mock tools

A simulation that reaches your real tools has real side effects: it books the appointment, sends the message, charges the card. And a real backend only ever shows you the branch its data happens to be on — "the calendar is full", "the lookup fails", "the booking API errors" are where voice agents die in production, and none of them can be ordered up from a real backend on demand.

A mock tool answers for one of your agent's tools during a simulation. It is written on the test itself, beside that test's scenario and expected behaviors, and matched strictly by tool name: one answer, or one error, per tool per test. This package is the piece that lives in your own agent's process and lets Egma answer.

Egma answers for exactly the tools the running test names. Every other tool runs its real implementation, untouched, and Egma is not in that path. A test that names none runs your agent against your real backend from end to end.

Unmocked tools run real, and are recorded from the agent's POV. The record of a simulation is your agent's own account of the conversation, so every call it made is on it, with the arguments the model sent and the result it received. A call a mock tool answered is marked mocked, beside the tool's own name; a call Egma refused shows as the error this package raised on it; a call carrying neither is a real one that ran.

A mock tool whose name never matches one of your agent's tools runs nothing and leaves no trace: the model never calls that name, and the record shows no call and no warning. Spell each tool exactly as your agent registers it.

Use

One call, after the agent is built and before the session starts:

from egma import simulation

async def entrypoint(ctx: agents.JobContext) -> None:
    agent = Agent(instructions=INSTRUCTIONS, tools=[check_calendar, book_appointment])
    session = AgentSession(stt=..., llm=..., tts=...)

    await simulation(agent, ctx, session)

    await session.start(agent=agent, room=ctx.room)

That is the whole simulation integration. It reads EGMA_URL and EGMA_API_KEY, the same two settings monitor reads, or the matching endpoint= and api_key= arguments.

One LiveKit job per process. The room a process exports under is fixed when its exporter is built and cannot be rewritten, so a second job in the same process is refused rather than filed under the first job's room. LiveKit runs one job per process by default; keep it that way.

What it sends to Egma

Your agent's own spans, over OTLP, to EGMA_URL with your project API key — the same road monitor uses. Each span carries the room's name, so Egma files them under the simulation that opened that room. They are batched at one second and flushed when the session closes and again when the LiveKit job stops, so the end of a conversation lands in Egma within a second or two of the caller leaving.

That is the agent's POV of the simulation: its turns, its tool calls, its own timings. Egma stores it beside what its own caller heard and shows it on the simulation.

If your worker already has an OpenTelemetry tracer provider — Langfuse, or your own collector — it is kept and used as it stands. Egma is added beside what you already export, never in place of it.

How it knows it is in a simulation

It reads the room's name off the job. Every room Egma conducts a simulation in is named egma-sim-…, and that prefix is fixed. Nothing else is read, nothing is asked, and no room is connected to find out.

The name is the signal because it is the one that arrives on every dispatch path an agent can end up in an Egma room by — whether Egma dispatches your worker by name, whether LiveKit walks an unnamed worker into the room, and whether your own token endpoint puts the agent there. A signal carried by an explicit dispatch arrives on only the first of those.

In a simulation room, simulation connects the job with LiveKit's own JobContext.connect() if your startup has not already done so, then finds Egma among the room's participants: Egma joins as egma-persona or egma-persona-<simulation>. On two of the three dispatch paths your agent is in the room before Egma, so it waits for that participant, for up to 45 seconds and without polling anything outside the room. That is a long time to hold an agent before it greets anybody, and it is the price of the wait being correct on every dispatch path rather than on one; a simulation ordinarily pays a fraction of it. It does not reconnect an already-connected room, and it never connects a production room.

If nobody by that name arrives, simulation raises egma.NotReported and your session never starts. If two participants answer to that name, the exchange is refused for the same reason a room with two claimants has no knowable answer — your tool inventory is not sent to either of them, and the same error is raised.

That is the fail-closed rule, and it is deliberate: a simulation that ran without reaching Egma would have called your real backends everywhere a mock tool was meant to answer, and its record would say nothing about it. A test that isolated nothing must not be allowed to look like one that did.

Then simulation reports your agent's tools to Egma — names and schemas, read off the agent object, so mock authoring starts from your real tool names instead of your memory of them — and asks which tools this simulation answers for. Egma replies with exactly the names the running test wrote, and couriers go in front of those names only. Calls to them go to Egma and come back with the authored answer. Every one of them lands on the simulation's record with its arguments, its answer, how long it took, and which mock tool answered.

In every other room it does nothing at all. A room your own system named — which is every production room — is a room where simulation returns having touched nothing: no wrapper, no message, no exporter, no connect. Your tools are the same objects, called the same way, with no wrapper between them and the model. Zero added latency, by construction rather than by care. That property is a test in this package (tests/test_inert.py), not a promise in this file.

Your job's dispatch metadata carries the test's own env. This SDK writes nothing into it and reads nothing out of it — not one key, in any room, for any purpose. With Project credentials, Egma writes the running test's job_dispatch_metadata directly to the dispatch. With a token endpoint, Egma sends the same value in room_config and the endpoint copies that configuration into the token it mints. Either way, the worker receives one compact JSON string, key for key, and nothing of Egma's beside it; json.loads(ctx.job.metadata)["tenant"] reads the value that scenario meant it to read. A test that names no env dispatches the empty string. The room's metadata Egma always leaves empty.

Where to call it

After the agent object exists and before session.start. The report of your tools is the first thing said, so an Egma that is not in the room is found before any tool call rather than half way through a test.

Keep one simulation call for the initial agent. The SDK follows LiveKit's public handoff events and installs the same simulation couriers for each selected Agent or AgentTask before that activity starts, so a tool the test names is answered whichever agent holds it.

What a call to a mocked tool does

• Goes to Egma over the same LiveKit room. No new endpoint, no new credential, nothing new to expose. If needed, simulation connects that room through the job context before sending the first RPC.
• Comes back with the authored answer, or raises the authored error as the tool's own error, so your agent handles it exactly as it would handle a real backend failing.
• Fails, if Egma turns out not to be reachable. The call raises as the tool's own error and your real tool is never run. A mocked tool exists because this test answers for it, so running its real implementation would book the real appointment and charge the real card — which is the one thing a test may never do.
• Never waits forever. Every branch ends in an answer or an error the model can hear.

A tool you attach to the agent after calling simulation is still intercepted on its first call — Egma's answers are held by name. Its arguments may be incomplete on the record, and Egma marks that call so you can see it.

Logging

Everything this package says goes to the egma logger. It is worth having on at INFO the first time you wire an agent up: the line after the tool report names how many tools you have and how many Egma answers for.

A simulation that could not report to Egma raises egma.NotReported rather than logging: no Egma participant arrived in the room, two participants claimed to be Egma, the room would not open, or the two halves do not speak the same version of the exchange. The message names what happened and the two things to check — the room and Egma's own side of it, then the package installed here. None of that is reachable in a production room.

Before you install anything: the interim recipe

You can get isolation today with no Egma code at all, using LiveKit's own mock_tools and a guard you write yourself:

from livekit.agents import mock_tools


def in_a_simulation(ctx: agents.JobContext) -> bool:
    return ctx.job.room.name.startswith("egma-sim-")


async def entrypoint(ctx: agents.JobContext) -> None:
    await ctx.connect()
    agent = Agent(instructions=INSTRUCTIONS, tools=[check_calendar])
    session = AgentSession(stt=..., llm=..., tts=...)

    if in_a_simulation(ctx):
        mock_tools(
            type(agent),
            {"check_calendar": lambda day: "No free slots on that day."},
            session=session,
        )

    await session.start(agent=agent, room=ctx.room)

A room your own system named means the guard is false and nothing is wrapped. Note where that safety comes from, because it is not where simulation's comes from: this guard fires on a prefix being present rather than on Egma being absent. So it is true in any room whose name begins egma-sim-, including one Egma is not in — simulation waits for Egma's participant and then fails the simulation, while this guard has nobody to wait for and simply mocks. Refuse that prefix wherever your own side mints production tokens, or a production room named to look like a simulation runs your canned answers against a live caller.

Write the guard on the room name, and not on egmaIdentity in ctx.job.metadata. That key is in no simulation room at all: the dispatch carries the test's own env and no key of Egma's. A guard on it does not raise — it quietly fails to fire, and every real tool it was meant to hold back runs inside a simulation.

What it cannot do is the rest of the job. One canned world for every test, so you cannot write "the calendar is full" as a test — you would be editing your agent's source to change a test's data. Nothing about those calls reaches Egma's record: no arguments, no answers, no timings, so graders that read tool facts have nothing to read. And it sends no spans, so the simulation has no agent POV — which a LiveKit simulation now requires, so a run set up this way fails.

And the honest caveat: that guard couples your agent's source to how Egma announces itself. The room-name prefix is a stated contract rather than an implementation detail, so it is the safe thing to key off — but the rest of the mechanism is not: where Egma sits in the room, what it is called, how long it takes to arrive, and what a room with two claimants means are all Egma's to evolve. That is precisely what simulation exists to own, and your side stays one line.

Use the recipe as the bridge, not as the small tier.

Upgrading from 0.2

Version 0.3 replaces mockable with simulation and monitor_livekit with monitor. Update the imports and calls; the old names are no longer exported.

simulation now also exports the agent's traces. Set EGMA_URL and EGMA_API_KEY, or pass endpoint and api_key, before calling it. A simulation that cannot report its tools raises NotReported before the session starts. The supported LiveKit range remains >=1.6.7,<1.7.

Compatibility

The egma-sim- room-name prefix is a stated contract, not an internal detail. Every room Egma opens for a simulation begins with it, on every dispatch path, and it is there to be relied on and to be allowlisted in a token endpoint. This SDK keys its whole simulation/production decision off it.

Every Egma deployment names its rooms that way, so this package works against a self-hosted Egma on whatever schedule its owner upgrades it. Nothing else is consulted — in particular, neither metadata channel, which carries your own configured JSON and nothing of Egma's.

This package pins livekit-agents to one minor version (>=1.6.7,<1.7), and that is deliberate. Interception uses LiveKit's testing API. Monitoring also reads LiveKit's current dynamic tracer provider because the public telemetry API has a setter but no getter. The fixture and live tests verify both seams before the supported range changes.

CI builds the wheel once and runs the core SDK tests against the minimum supported LiveKit version and the version in uv.lock, in parallel clean environments. The live tests run once on the locked version after those checks, so they have sole use of the local LiveKit server port.

The package also keeps the OpenAI Python package on major version 2. LiveKit Agents 1.6 does not support OpenAI 3.

mock_tools writes into a side table LiveKit keeps per session. It does not touch your agent, your agent's class, or your tool registry, and the model keeps seeing your real tool schemas throughout. Never calling it leaves everything byte for byte as it was, which is what makes the inert path literal.

If that API ever moves, the documented fallback is Agent.update_tools(...) with function_tool(raw_schema=...) wrappers built from the real tools' schemas. It is a heavier mechanism — it takes responsibility for schema fidelity, for the real implementations, and for re-wrapping after a handoff — which is exactly why it is the fallback and not the mechanism.

Developing this package

The toolchain is uv.

uv sync
uv run ruff check src tests
uv run pytest

The normal suite is hermetic: no LiveKit server, no project, and no external network. One monitoring test uses a local HTTP collector. One live test skips visibly when the environment is silent, naming what it needs. To run it, point it at a LiveKit project:

TEST_LIVEKIT_URL=wss://... \
TEST_LIVEKIT_API_KEY=... TEST_LIVEKIT_API_SECRET=... \
TEST_MODEL_API_KEY=... \
uv run pytest tests/test_live_mockable.py -v

Each name falls back to the plain one the tool's own CLI reads — LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET, OPENAI_API_KEY — so one environment serves this and whatever else you run beside it. TEST_MODEL_NAME picks the model; it defaults to gpt-4o-mini.

License

Apache-2.0, with the rest of Egma.