🏁 Getting Started with the Agent SDK
This guide walks you through creating your first custom agent using the STELLA Agent SDK.
Prerequisites
- Python 3.9+
- A running STELLA backend
- LiveKit server access
Installation
pip install stella-agent-sdk
Project Setup
Create a new directory for your agent:
mkdir my-custom-agent
cd my-custom-agent
Create the project structure:
my-custom-agent/
├── agent.py # Main agent code
├── requirements.txt # Dependencies
├── Dockerfile # Container configuration
└── config.yaml # Agent configuration
Your First Agent
Create agent.py:
import asyncio
from stella_sdk import BaseAgent, AudioPipeline
class MyFirstAgent(BaseAgent):
"""A simple agent that responds to user messages."""
def __init__(self):
super().__init__()
self.pipeline = AudioPipeline()
self.conversation_history = []
async def on_connect(self):
"""Called when agent connects to the room."""
print(f"Connected to room: {self.room_name}")
await self.send_status("ready")
# Optional: Send a greeting
greeting = "Hello! I'm your AI assistant. How can I help you?"
await self.speak(greeting)
async def on_disconnect(self):
"""Called when agent disconnects."""
print("Disconnected from room")
async def on_transcript(self, text: str, is_final: bool):
"""Called when user speech is transcribed."""
if not is_final:
# Partial transcript - could update UI
return
# Final transcript - generate response
print(f"User said: {text}")
# Add to conversation history
self.conversation_history.append({
"role": "user",
"content": text
})
# Generate and speak response
response = await self.generate_response(text)
await self.speak(response)
# Add response to history
self.conversation_history.append({
"role": "assistant",
"content": response
})
async def on_data_message(self, message: dict):
"""Called when receiving a data channel message."""
if message.get("type") == "user_text":
# Handle text input
await self.on_transcript(message["data"], is_final=True)
async def generate_response(self, user_input: str) -> str:
"""Generate a response using OpenAI."""
from openai import AsyncOpenAI
client = AsyncOpenAI()
messages = [
{"role": "system", "content": "You are a helpful assistant."},
*self.conversation_history
]
response = await client.chat.completions.create(
model="gpt-4o",
messages=messages,
stream=False
)
return response.choices[0].message.content
async def speak(self, text: str):
"""Convert text to speech and publish to room."""
await self.send_transcript(text, speaker="assistant")
audio = await self.pipeline.text_to_speech(text)
await self.publish_audio(audio)
if __name__ == "__main__":
agent = MyFirstAgent()
agent.run()
Configuration
Create config.yaml:
agent:
name: my-first-agent
version: 1.0.0
livekit:
url: ${LIVEKIT_URL}
api_key: ${LIVEKIT_API_KEY}
api_secret: ${LIVEKIT_API_SECRET}
openai:
api_key: ${OPENAI_API_KEY}
model: gpt-4o
audio:
stt_provider: sherpa
tts_provider: kokoro
Dependencies
Create requirements.txt:
stella-agent-sdk>=1.0.0
openai>=1.0.0
livekit>=0.10.0
Dockerfile
Create Dockerfile:
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy agent code
COPY . .
# Run the agent
CMD ["python", "agent.py"]
Running Locally
Set environment variables:
export LIVEKIT_URL=ws://localhost:7880
export LIVEKIT_API_KEY=devkey
export LIVEKIT_API_SECRET=secret
export OPENAI_API_KEY=sk-...
export ROOM_NAME=test-room
export PARTICIPANT_IDENTITY=my-agent
Run the agent:
python agent.py
Building and Deploying
Build the Docker image:
docker build -t my-custom-agent:latest .
Deploy to STELLA by updating the agent image configuration:
# In your .env or ConfigMap
AGENT_IMAGE=my-custom-agent:latest
Testing Your Agent
- Start STELLA:
./scripts/start-k8s.sh - Create a session in the Frontend UI
- Deploy your custom agent
- Interact via voice or text
Next Steps
- Base Agent - Learn about all BaseAgent methods
- Audio Pipeline - Configure STT/TTS
- Message Types - Understand the message protocol
- Tools - Add custom tool support