Egma Python SDK: Mock Tools for LiveKit Voice Agents

​

Install

pip install egma

uv add egma

​

The one API: mockable

from egma import mockable

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

​

What mockable does

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.

​

Placement

​

Production safety guarantee

• 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.

​

How tool matching works

​

Logging

import logging
logging.getLogger("egma").setLevel(logging.INFO)

​

Version pinning

​

Before you install anything: the interim recipe

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)

• 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.