Open-source Python framework for building production-ready, real-time voice and multimodal AI agents.
The VideoSDK AI Agents framework is a Python SDK for building AI agents that join VideoSDK rooms as real-time participants. It connects your agent worker, AI models, and user devices into a single low-latency pipeline β handling audio streaming, turn detection, interruptions, and media routing automatically so you can focus on agent logic.
VideoSDK AI Agents is a Python framework that lets you build voice and multimodal AI agents that participate directly in VideoSDK rooms. The framework manages the full agent lifecycle β from joining a room and processing live audio, to running STT β LLM β TTS pipelines or connecting to unified realtime models, to handling turn detection, VAD, interruptions, and clean teardown.
v1.0.0 introduces a unified Pipeline class that replaces the previous CascadingPipeline and RealtimePipeline. Pass in any combination of components β STT, LLM, TTS, VAD, turn detector, avatar β and the framework wires them together and selects the optimal execution mode automatically. A decorator-based hooks system (@pipeline.on(...)) lets you intercept and transform data at any stage without subclassing.
ποΈ Agent with Cascade ModeBuild an AI Voice Agent using Cascade Mode (STT β LLM β TTS). |
Build an AI Voice Agent using a unified Realtime model (e.g. Gemini Live). |
π» Agent DocumentationThe VideoSDK Agent Official Documentation. |
π SDK ReferenceReference Docs for Agents Framework. |
| # | Feature | Description |
|---|---|---|
| 1 | π€ Real-time Communication (Audio/Video) | Agents can listen, speak, and interact live in meetings. |
| 2 | π SIP & Telephony Integration | Seamlessly connect agents to phone systems via SIP for call handling, routing, and PSTN access. |
| 3 | π§ Virtual Avatars | Build or plug in any avatar provider β the framework handles audio routing, sync, and teardown automatically. |
| 4 | π€ Multi-Model Support | Integrate with OpenAI, Gemini, AWS NovaSonic, Anthropic, and more. |
| 5 | π§© Cascade Mode | Compose any STT β LLM β TTS chain across providers for full control and flexibility. |
| 6 | β‘ Realtime Mode | Use unified realtime models (OpenAI Realtime, AWS Nova Sonic, Gemini Live) for lowest latency. |
| 7 | π Hybrid Mode | Mix cascade and realtime components β custom STT with a realtime model, or realtime with custom TTS. |
| 8 | πͺ Pipeline Hooks | Intercept and transform data at any stage (STT, LLM, TTS, turns) using @pipeline.on(...). |
| 9 | π οΈ Function Tools | Extend agent capabilities with any external tool or API call. |
| 10 | π MCP Integration | Connect agents to external data sources and tools using Model Context Protocol. |
| 11 | π A2A Protocol | Reliable agent-to-agent routing with correlation-based request tracking. |
| 12 | π¦ LangChain & LangGraph | Plug in any LangChain BaseChatModel or LangGraph StateGraph as the agent's LLM. |
| 13 | π Observability | Built-in metrics, OpenTelemetry tracing, and structured logging per component. |
Important
Star VideoSDK Repositories βοΈ
Get instant notifications for new releases and updates. Your support helps us grow and improve VideoSDK!
All agents are built around a single Pipeline class. Pass in your components β the SDK picks the right execution mode automatically.
Mix and match any provider for each stage. Best when you need custom STT, specific LLM behaviour, or a particular TTS voice.
async def start_session(context: JobContext):
pipeline = Pipeline(
stt=DeepgramSTT(),
llm=GoogleLLM(),
tts=CartesiaTTS(),
vad=SileroVAD(),
turn_detector=TurnDetector(),
)
session = AgentSession(agent=MyAgent(), pipeline=pipeline)
await session.start(wait_for_participant=True, run_until_shutdown=True)Use a single realtime model for the entire voice pipeline. Best for sub-500ms response latency.
async def start_session(context: JobContext):
pipeline = Pipeline(
llm=GeminiRealtime(
model="gemini-3.1-flash-live-preview",
config=GeminiLiveConfig(voice="Leda", response_modalities=["AUDIO"]),
)
)
session = AgentSession(agent=MyAgent(), pipeline=pipeline)
await session.start(wait_for_participant=True, run_until_shutdown=True)Use an external STT with a Realtime LLM, or a Realtime model with a custom TTS:
# External STT β Realtime LLM
pipeline = Pipeline(stt=DeepgramSTT(), llm=OpenAIRealtime(...))
# Realtime LLM β External TTS
pipeline = Pipeline(llm=OpenAIRealtime(...), tts=ElevenLabsTTS(...))@pipeline.on("stt")
async def clean_transcript(text: str) -> str:
return text.strip()
@pipeline.on("llm")
async def route_llm(messages):
if "transfer" in messages[-1].content:
yield "Transferring you now." # bypass LLM entirely
@pipeline.on("tts")
async def fix_pronunciation(text: str) -> str:
return text.replace("VideoSDK", "Video S D K")
@pipeline.on("user_turn_start")
async def on_user_starts():
print("User is speaking...")Available hook points: stt Β· tts Β· llm Β· vision_frame Β· user_turn_start Β· user_turn_end Β· agent_turn_start Β· agent_turn_end
Before you begin, ensure you have:
- A VideoSDK authentication token (generate from app.videosdk.live)
- A VideoSDK meeting ID (you can generate one using the Create Room API or through the VideoSDK dashboard)
- Python 3.12 or higher
- Third-Party API Keys:
- API keys for the services you intend to use (e.g., OpenAI for LLM/STT/TTS, ElevenLabs for TTS, Google for Gemini etc.).
UV is a fast Python package manager that handles virtual environments and dependency management automatically.
If you don't have UV installed, see the UV installation guide.
-
Install the core VideoSDK AI Agent package:
uv add videosdk-agents
-
Install Optional Plugins:
uv add videosdk-plugins-openai uv add videosdk-plugins-deepgram
-
Run your agent:
uv run python main.py
-
Create and activate a virtual environment with Python 3.12 or higher.
macOS / Linux
python3 -m venv venv source venv/bin/activateWindows
python -m venv venv venv\Scripts\activate
-
Install the core VideoSDK AI Agent package
pip install videosdk-agents
-
Install Optional Plugins. Plugins help integrate different providers for Realtime, STT, LLM, TTS, and more. Install what your use case needs:
# Example: Install the Turn Detector plugin pip install videosdk-plugins-turn-detectorπ Supported plugins (Realtime, LLM, STT, TTS, VAD, Avatar, SIP) are listed in the Supported Libraries section below.
To set up the project locally, clone the repo and install all packages (core + all plugins) as editable installs:
Using UV (Recommended):
git clone https://github.com/videosdk-live/agents.git
cd agents
uv sync
uv run python examples/cascade_basic.pyUsing pip:
git clone https://github.com/videosdk-live/agents.git
cd agents
bash setup.sh
source venv/bin/activate
python examples/cascade_basic.pyBefore your AI agent can join a meeting, you'll need to create a meeting ID. You can generate one using the VideoSDK Create Room API:
curl -X POST https://api.videosdk.live/v2/rooms \
-H "Authorization: YOUR_JWT_TOKEN_HERE" \
-H "Content-Type: application/json"For more details on the Create Room API, refer to the VideoSDK documentation.
Now that you've installed the necessary packages, you're ready to build!
First, let's create a custom voice agent by inheriting from the base Agent class:
from videosdk.agents import Agent, function_tool
# External Tool
# async def get_weather(self, latitude: str, longitude: str):
class VoiceAgent(Agent):
def __init__(self):
super().__init__(
instructions="You are a helpful voice assistant that can answer questions and help with tasks.",
tools=[get_weather] # You can register any external tool defined outside of this scope
)
async def on_enter(self) -> None:
"""Called when the agent first joins the meeting"""
await self.session.say("Hi there! How can I help you today?")
async def on_exit(self) -> None:
"""Called when the agent exits the meeting"""
await self.session.say("Goodbye!")This code defines a basic voice agent with:
- Custom instructions that define the agent's personality and capabilities
- An entry message when joining a meeting
- State change handling to track the agent's current activity
Function tools allow your agent to perform actions beyond conversation. There are two ways to define tools:
- External Tools: Defined as standalone functions outside the agent class and registered via the
toolsargument in the agent's constructor. - Internal Tools: Defined as methods inside the agent class and decorated with
@function_tool.
Below is an example of both:
import aiohttp
# External Function Tools
@function_tool
def get_weather(latitude: str, longitude: str):
print(f"Getting weather for {latitude}, {longitude}")
url = f"https://api.open-meteo.com/v1/forecast?latitude={latitude}&longitude={longitude}¤t=temperature_2m"
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
data = await response.json()
return {
"temperature": data["current"]["temperature_2m"],
"temperature_unit": "Celsius",
}
else:
raise Exception(
f"Failed to get weather data, status code: {response.status}"
)
class VoiceAgent(Agent):
# ... previous code ...
# Internal Function Tools
@function_tool
async def get_horoscope(self, sign: str) -> dict:
horoscopes = {
"Aries": "Today is your lucky day!",
"Taurus": "Focus on your goals today.",
"Gemini": "Communication will be important today.",
}
return {
"sign": sign,
"horoscope": horoscopes.get(sign, "The stars are aligned for you today!"),
}- Use external tools for reusable, standalone functions (registered via
tools=[...]). - Use internal tools for agent-specific logic as class methods.
- Both must be decorated with
@function_toolfor the agent to recognize and use them.
Connect your agent to an AI model using the unified Pipeline class. Pass in whichever components you need β the SDK handles the rest.
Realtime mode (single model, lowest latency):
async def start_session(context: JobContext):
pipeline = Pipeline(
llm=GeminiRealtime(
model="gemini-3.1-flash-live-preview",
config=GeminiLiveConfig(voice="Leda", response_modalities=["AUDIO"]),
)
)
session = AgentSession(agent=VoiceAgent(), pipeline=pipeline)
await session.start(wait_for_participant=True, run_until_shutdown=True)Cascade mode (STT β LLM β TTS, full provider control):
async def start_session(context: JobContext):
pipeline = Pipeline(
stt=DeepgramSTT(),
llm=GoogleLLM(),
tts=CartesiaTTS(),
vad=SileroVAD(),
turn_detector=TurnDetector(),
)
session = AgentSession(agent=VoiceAgent(), pipeline=pipeline)
await session.start(wait_for_participant=True, run_until_shutdown=True)from videosdk.agents import AgentSession, WorkerJob, RoomOptions, JobContext
async def start_session(context: JobContext):
session = AgentSession(
agent=VoiceAgent(),
pipeline=pipeline,
)
await session.start(wait_for_participant=True, run_until_shutdown=True)
def make_context() -> JobContext:
room_options = RoomOptions(
room_id="<meeting_id>",
name="Test Agent",
playground=True,
)
return JobContext(room_options=room_options)
if __name__ == "__main__":
job = WorkerJob(entrypoint=start_session, jobctx=make_context)
job.start()After setting up your AI Agent, you'll need a client application to connect with it. You can use any of the VideoSDK quickstart examples to create a client that joins the same meeting:
When setting up your client application, make sure to use the same meeting ID that your AI Agent is using.
Once you have completed the setup, you can run your AI Voice Agent project using Python. Make sure your .env file is properly configured and all dependencies are installed.
python main.pyTip
Console Mode β test your agent locally without a meeting room.
Set playground=True in RoomOptions and run python main.py to interact via your mic and speakers directly from the terminal.
For deployment options and guide, checkout the official documentation here: Deployment
VideoSDK Inference provides a unified gateway to access STT, LLM, TTS, Denoise, and Realtime models β without managing individual provider API keys. Authentication is handled via your VIDEOSDK_AUTH_TOKEN and usage is billed from your VideoSDK account balance.
from videosdk.agents.inference import STT, LLM, TTS, Denoise, RealtimeCascade Mode with VideoSDK Inference:
async def start_session(context: JobContext):
pipeline = Pipeline(
stt=STT.sarvam(model_id="saarika:v2.5", language="en-IN"),
llm=LLM.google(model_id="gemini-2.5-flash"),
tts=TTS.sarvam(model_id="bulbul:v2", speaker="anushka", language="en-IN"),
denoise=Denoise.sanas(),
vad=SileroVAD(),
)
session = AgentSession(agent=MyAgent(), pipeline=pipeline)
await session.start(wait_for_participant=True, run_until_shutdown=True)Realtime Mode with VideoSDK Inference:
async def start_session(context: JobContext):
pipeline = Pipeline(
llm=Realtime.gemini(
model_id="gemini-3.1-flash-live-preview",
voice="Puck",
language_code="en-US",
response_modalities=["AUDIO"],
)
)
session = AgentSession(agent=MyAgent(), pipeline=pipeline)
await session.start(wait_for_participant=True, run_until_shutdown=True)See Inference Pricing for provider-wise billing details.
The framework supports integration with various AI models and tools, across multiple categories:
| Category | Services |
|---|---|
| Real-time Models | OpenAI | Gemini | AWS Nova Sonic | Azure Voice Live |
| Speech-to-Text (STT) | OpenAI | Google | Azure AI Speech | Azure OpenAI | Sarvam AI | Deepgram | Cartesia | AssemblyAI | Navana |
| Language Models (LLM) | OpenAI | Azure OpenAI | Google | Sarvam AI | Anthropic | Cerebras |
| Text-to-Speech (TTS) | OpenAI | Google | AWS Polly | Azure AI Speech | Azure OpenAI | Deepgram | Sarvam AI | ElevenLabs | Cartesia | Resemble AI | Smallest AI | Speechify | InWorld | Neuphonic | Rime AI | Hume AI | Groq | LMNT AI | Papla Media |
| Voice Activity Detection (VAD) | SileroVAD |
| Turn Detection Model | Namo Turn Detector |
| Virtual Avatar | Simli | Anam | Custom (implement connect / aclose protocol) |
| LLM Orchestration | LangChain | LangGraph |
| Denoise | RNNoise |
Tip
Installation Examples
# Install with specific plugins
pip install videosdk-agents[openai,elevenlabs,silero]
# Install individual plugins
pip install videosdk-plugins-anthropic
pip install videosdk-plugins-deepgramExplore the following examples to see the framework in action:
ποΈ Cascade Mode (Basic)Simple STT β LLM β TTS voice agent using Google LLM + Deepgram STT + Cartesia TTS. |
Advanced cascade agent with VAD, turn detection, and interruption handling. |
β‘ Realtime ModeMinimal realtime agent using Gemini Live for lowest-latency voice interactions. |
π Hybrid ModeMix cascade and realtime β custom STT with a realtime model, or realtime with custom TTS. |
π§© Composable PipelinesFlexible Pipeline configs β transcription-only, LLM-only, voice+chat, full voice agent. |
πͺ Pipeline HooksIntercept and transform STT, LLM, and TTS data at any stage using |
|
Stock Market Analyst Agent with real-time market data access via Model Context Protocol. |
π€ Agent-to-Agent (A2A)Multi-agent workflow: customer agent that transfers loan queries to a Loan Specialist Agent. |
|
Use LangChain tools and agents within the VideoSDK agent framework. |
πΈοΈ LangGraph IntegrationOrchestrate multi-step agent workflows using LangGraph state machines. |
|
Persistent memory across sessions using Mem0 for long-term context retention. |
ποΈ Vision AgentMultimodal agent that processes video frames alongside voice using cascading or realtime pipelines. |
|
Trigger n8n automation workflows from within your agent using webhooks. |
π§βπΌ Human in the LoopEscalate to a human agent mid-conversation via Discord or other channels. |
π AI Telephony AgentHospital appointment booking via a voice-enabled telephony agent. |
|
|
Agent that answers questions based on documentation knowledge. |
π Virtual Avatar AgentA Virtual Avatar Agent that presents a weather forecast. |
π₯ Appointment BookingHealthcare front-desk receptionist for scheduling clinic appointments. |
π£ Announcement AgentProactive outbound agent for broadcasting announcements. |
π§ Customer SupportAI-powered customer support agent with escalation and knowledge base. |
π More Use CasesCall center, IVR, medical triage, language tutor, meeting notes, and more. |
For comprehensive guides and API references:
|
Complete framework documentation |
π API ReferenceDetailed API documentation |
π Examples DirectoryAdditional code examples |
We welcome contributions! Here's how you can help:
π Report IssuesOpen an issue for bugs or feature requests |
π Submit PRsCreate a pull request with improvements |
π οΈ Build PluginsFollow our plugin development guide |
π¬ Join CommunityConnect with us on Discord |
The framework is under active development, so contributions in the form of new plugins, features, bug fixes, or documentation improvements are highly appreciated.
Want to integrate a new AI provider? Check out BUILD YOUR OWN PLUGIN for:
- Step-by-step plugin creation guide
- Directory structure and file requirements
- Implementation examples for STT, LLM, and TTS
- Testing and submission guidelines
Stay connected with VideoSDK:
π¬ DiscordJoin our community |
π¦ Twitter@video_sdk |
|
π LinkedInVideoSDK Company |
Tip
Support the Project! βοΈ
Star the repository, join the community, and help us improve VideoSDK by providing feedback, reporting bugs, or contributing plugins.
Made with β€οΈ by The VideoSDK Team