> ## Documentation Index
> Fetch the complete documentation index at: https://docs.egma.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Egma Python SDK: Mock Tools for LiveKit Voice Agents

> Add one line to your LiveKit agent entrypoint and Egma intercepts tool calls during simulations — zero impact on production behavior, guaranteed.

The `egma` Python package lets Egma answer your agent's tool calls during a simulation, without touching anything in production. Add one call after your agent is built and Egma intercepts the tools it has mock answers for. Every other tool — and every room with no Egma in it — runs exactly as if the package were not installed.

This solves a real problem: 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, but none of them can be ordered up on demand from a real backend.

## Install

```bash theme={null}
pip install egma
```

Or with uv:

```bash theme={null}
uv add egma
```

Requires Python 3.11 or newer and `livekit-agents>=1.6.7,<1.7`.

<Note>
  The `livekit-agents` version is pinned to one minor version intentionally. The interception mechanism uses LiveKit's own `mock_tools`, which lives in the framework's testing namespace and carries no stability promise. The pin plus a live smoke test — a real session in a real room — is how this package verifies the mechanism still works before you find out in a simulation. See [Version pinning](#version-pinning) below.
</Note>

## The one API: `mockable`

Import `mockable` and call it once in your agent's entrypoint, after the agent object exists and before `session.start`:

```python theme={null}
from egma import mockable
```

```python theme={null}
import asyncio
from livekit import agents
from livekit.agents import AgentSession, Agent, RoomInputOptions
from livekit.plugins import openai, silero

INSTRUCTIONS = "You are a helpful dental office receptionist..."

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

    agent = Agent(
        instructions=INSTRUCTIONS,
        tools=[check_calendar, book_appointment, lookup_patient],
    )
    session = AgentSession(
        stt=openai.STT(model="gpt-4o-mini-transcribe", use_realtime=True),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=openai.TTS(model="gpt-4o-mini-tts", voice="ash"),
        vad=silero.VAD.load(),
    )

    # Add this line. That is the whole integration.
    await mockable(agent, ctx, session)

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


if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))
```

That is the whole integration.

## What `mockable` does

`mockable` reads the dispatch metadata on `ctx.job` once. In any room without an Egma participant in it — which is every production room — it returns immediately having touched nothing. Your tools are the same objects, called the same way, with no wrapper between them and the model. This is enforced by a test in the package (`tests/test_inert.py`), not just described in documentation.

In a simulation, `mockable`:

1. **Reads your agent's tools** — names and schemas — off the `agent` object and sends them to Egma as a census. Mock authoring in Egma starts from your real tool names and signatures, not from your memory of them.
2. **Learns which tools Egma answers for** in this simulation.
3. **Installs a courier** in front of each of those tools using LiveKit's own `mock_tools`. Couriers go into a side table LiveKit keeps per session; they do not touch your agent, your agent's class, or your tool registry. The model keeps seeing your real tool schemas throughout.
4. **Leaves every other tool exactly as it was.**

Every covered tool call lands on the simulation record with its arguments, its answer, how long it took, and which mock tool answered it. Tools that ran their real implementation are also marked, so you always know whether a simulation was fully isolated.

### Placement

Call `mockable` after the agent object exists and before `session.start`. The census is the first message sent, so an Egma that is not in the room is discovered before any tool call rather than halfway through a test.

If your application hands off between several `Agent` classes, call `mockable` once per class you want covered.

## Production safety guarantee

The inert path is not a best-effort claim — it is a construction property with a test behind it.

* **No Egma participant in the room** → `mockable` returns immediately. Zero added latency, no side effects, no wrapped tools.
* **Egma present but unreachable mid-simulation** → the real tool runs. A room that lost its Egma participant behaves as a room that never had one.
* **Egma explicitly refuses a call** (unknown tool, malformed payload) → `ToolError` is raised so the model hears a tool that failed and can respond accordingly. Falling open on a refusal would mean a simulation has real side effects on Egma's behalf, which is never the right outcome.

| Situation                             | Behavior                               |
| ------------------------------------- | -------------------------------------- |
| Production room (no Egma)             | Returns immediately, nothing wrapped   |
| Simulation — tool has a mock          | Egma answers; result returned to model |
| Simulation — tool has no mock         | Real tool runs                         |
| Egma not reachable (fail-open)        | Real tool runs                         |
| Egma explicitly refuses (fail-closed) | `ToolError` raised to model            |

## How tool matching works

Mock tools are matched by tool name, strictly. Egma sends a list of the names it will answer for in this simulation. A courier is installed for each of those names, whether or not the agent currently has a tool by that name. Matching happens per call, by name, against the side table LiveKit keeps.

A tool you attach to the agent *after* calling `mockable` is still intercepted on its first call — because couriers are matched by name at call time, not at registration time. Its arguments may be incomplete on the record (since there was no signature to read them through when the courier was made), and Egma marks that call so you can see it.

The transport used for tool calls is LiveKit RPC over the room you are already connected to. No new endpoint, no new credential, nothing new to expose.

## Logging

Everything this package says goes to the `egma` logger:

```python theme={null}
import logging
logging.getLogger("egma").setLevel(logging.INFO)
```

At `INFO` level, the line after the census names how many tools your agent has and how many Egma answers for in this simulation. This is the first thing to check when wiring an agent up for the first time.

## Version pinning

The `livekit-agents` dependency is pinned to `>=1.6.7,<1.7`. This is deliberate, not an oversight.

The interception mechanism uses `mock_tools` from LiveKit's testing namespace. That API carries no stability promise. The pin means an unread minor version cannot arrive by itself and silently break interception. Raising the ceiling is a deliberate act with a live smoke test as evidence — a real session in a real room, proving that interception actually happened.

If `mock_tools` ever moves in a future version, the documented fallback is `Agent.update_tools(...)` with `function_tool(raw_schema=...)` wrappers built from the real tools' schemas. That is a heavier mechanism — it owns schema fidelity, the real implementations, and re-wrapping after handoffs — which is exactly why it is the fallback and not the current approach.

## Before you install anything: the interim recipe

If you need tool isolation today without the Egma SDK, you can use LiveKit's own `mock_tools` with a guard you write yourself:

```python theme={null}
import json
from livekit.agents import mock_tools


def in_a_simulation(ctx: agents.JobContext) -> bool:
    try:
        return bool(json.loads(ctx.job.metadata or "{}").get("egmaIdentity"))
    except ValueError:
        return False


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)
```

This is production-safe by the same logic — no Egma in the room means the guard is false and nothing is wrapped. But it has real limitations:

* One canned answer per tool for every simulation. You cannot author different answers for different tests without editing your agent's source code.
* Nothing about those tool calls reaches Egma's record: no arguments, no answers, no timings, no coverage stamp. Graders that read tool facts have nothing to read.
* No declared delay, so latency numbers from a mocked run will flatter you.
* The guard couples your agent to the exact shape of Egma's dispatch metadata today. If that shape changes, your agent breaks in a way your own tests would not catch. `mockable` exists to own that coupling so your side stays one line.

Use this as a bridge while migrating, not as a permanent solution.