Run your existing LiveKit agents across any text channel (like SMS, chat, or API) without modifying your agent code. LiveTxt extends LiveKit's real-time voice and video capabilities to text, enabling channel-agnostic AI agents for SMS and other messaging platforms. Enjoy drop-in compatibility with livekit-agents.
- Channel-agnostic by design (voice/video via LiveKit, text/API via LiveTxt)
- Drop-in compatibility with livekit-agents (no agent rewrites)
- Stateless text execution (each turn in → result out)
A single agents/multi-agent system API (livekit-agents) that runs on any channel.
- One codebase → many channels
- Works for single agents and coordinating multi-agent entrypoints
- LiveKit handles real-time voice/video; LiveTxt handles chat, SMS, and API
- Keep agent code identical; switch runtimes by channel
- Voice/Video: LiveKit room runtime
- Chat/SMS/API: LiveTxt
execute_job()
pip install -e .
# or with dev tools
pip install -e ".[dev]"The easiest way to run an agent is with the built-in HTTP server that auto-loads your agent:
# Start worker (agent auto-loads on startup)
python -m livetxt.cli serve examples/weather-agent/weather_agent.py --port 8081
# Agent is ready to receive requests - no manual loading needed!Then send requests from the gateway or directly:
curl -X POST http://localhost:8081/execute \
-H "Content-Type: application/json" \
-d '{
"request_id": "req_1",
"session_id": "session_1",
"user_id": "user_1",
"message": "What's the weather?"
}'Define your agent exactly as you do for livekit-agents, then execute it with a user message.
from livetxt import execute_job, JobRequest, SerializableSessionState
from livekit.agents import JobContext
async def my_agent(ctx: JobContext):
await ctx.connect()
@ctx.room.on("data_received")
def handle_message(data: bytes, topic: str, participant):
msg = data.decode("utf-8")
ctx.room.local_participant.publish_data(
f"You said: {msg}".encode("utf-8"),
topic="lk.chat",
)
request = JobRequest(
job_id="conversation_123",
user_input="Hello!",
state=SerializableSessionState(),
)
result = await execute_job(my_agent, request)
print(result.response_text) # You said: Hello!Pass the previous turn's state into the next request.
# turn 1
r1 = JobRequest(job_id="t1", user_input="My name is Alice", state=SerializableSessionState())
res1 = await execute_job(my_agent, r1)
# turn 2
r2 = JobRequest(job_id="t2", user_input="What's my name?", state=res1.updated_state)
res2 = await execute_job(my_agent, r2)execute_job()creates a text‑only JobContext with a fake Room and Participant.- Your agent registers handlers (e.g.,
data_received). - The user message is injected after handlers are registered.
- When your agent calls
publish_data(), the text is captured and returned as aJobResult.
- Single-agent and multi-agent entrypoints (multiple handlers/agents)
- Text agents using
ctx.room.on("data_received") - Multi‑turn conversations via
SerializableSessionState
Not yet:
- Agents built on
AgentSession+ LLM - Function/tool calling
- Audio/video (handled by LiveKit runtime)
See examples/ for runnable agents. A good place to start:
examples/weather-agent/
Each example folder includes its own README with run instructions.
# run all tests
pytest tests/ -v
# with coverage
pytest --cov=livetxt --cov-report=term tests/- No response? Ensure your handler listens to
data_receivedand you callawait ctx.connect()before relying on events. - Byte/str issues? Encode/decode UTF‑8 around
publish_data()and handler inputs. - State not sticking? Always pass
result.updated_stateinto the nextJobRequest.
Contributions are welcome! To set up for development:
# Install with dev dependencies
pip install -e ".[dev]"
# Run tests
pytest tests/ -v
# Run linting
ruff check --fix livetxt/
black livetxt/
mypy livetxt/See tests/README.md for testing guidelines and WARP.md for architecture notes.
Apache‑2.0. See LICENSE and NOTICE.
- livekit-agents: https://github.com/livekit/agents