Build Your Own Agent
The Agent SDK lets you focus purely on your agent's logic. STELLA handles the infrastructureβaudio pipeline, WebRTC streaming, session lifecycle, and deployment orchestrationβso you can concentrate on what makes your agent unique.
What STELLA handles:
- Audio pipeline (STT and TTS)
- WebRTC streaming via LiveKit
- Session creation, state management, and cleanup
- Deployment and scaling
- Message recording and transcript storage
What you implement:
- Conversation logic and prompts
- Tool calls and integrations
- Business rules and workflows
By the end of this guide, you'll have a working agent that can handle voice conversations, use custom tools, and integrate with your own services.
Overviewβ
STELLA agents are Python applications that:
- Connect to LiveKit rooms for real-time audio
- Process speech-to-text (STT) and text-to-speech (TTS)
- Use LLMs to generate intelligent responses
- Execute custom tools to interact with external systems
Project Setupβ
Create the project structure
Set up your agent project with the following structure:
my-agent/
βββ src/
β βββ __init__.py
β βββ agent.py # Main agent class
β βββ tools.py # Custom tools
β βββ config.py # Configuration
βββ Dockerfile
βββ requirements.txt
βββ .env
mkdir -p my-agent/src
cd my-agent
touch src/__init__.py src/agent.py src/tools.py src/config.py
touch Dockerfile requirements.txt .env
Define your dependencies
Create your requirements.txt with the necessary packages:
stella-agent-sdk>=1.0.0
openai>=1.0.0
livekit>=0.10.0
pydantic-settings>=2.0.0
Configure the agent
Create your configuration in src/config.py:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
# LiveKit connection
livekit_url: str
livekit_api_key: str
livekit_api_secret: str
# OpenAI
openai_api_key: str
openai_model: str = "gpt-4o"
# Audio pipeline
stt_provider: str = "sherpa"
tts_provider: str = "kokoro"
# Session
room_name: str
participant_identity: str = "agent"
class Config:
env_file = ".env"
settings = Settings()
Implement the agent
Create your main agent class in src/agent.py:
from openai import AsyncOpenAI
from stella_sdk import BaseAgent, AudioPipeline
from .config import settings
class MyCustomAgent(BaseAgent):
"""A custom conversational AI agent."""
def __init__(self):
super().__init__()
self.pipeline = AudioPipeline(
stt_provider=settings.stt_provider,
tts_provider=settings.tts_provider
)
self.openai = AsyncOpenAI()
self.history = []
async def on_connect(self):
"""Called when the agent connects to the LiveKit room."""
print(f"Connected to room: {self.room_name}")
# Send a greeting
greeting = "Hello! How can I help you today?"
await self.speak(greeting)
async def on_transcript(self, text: str, is_final: bool):
"""Handle transcribed user speech."""
if not is_final:
return
# Add to history
self.history.append({"role": "user", "content": text})
# Generate response
response = await self.generate_response(text)
# Speak the response
await self.speak(response)
# Add to history
self.history.append({"role": "assistant", "content": response})
async def generate_response(self, user_input: str) -> str:
"""Generate a response using OpenAI."""
response = await self.openai.chat.completions.create(
model=settings.openai_model,
messages=[
{"role": "system", "content": "You are a helpful assistant."},
*self.history
]
)
return response.choices[0].message.content
async def speak(self, text: str):
"""Convert text to speech and publish to the room."""
async for audio_chunk in self.pipeline.text_to_speech_stream(text):
await self.publish_audio(audio_chunk)
def main():
agent = MyCustomAgent()
agent.run()
if __name__ == "__main__":
main()
Adding Custom Toolsβ
Tools allow your agent to perform actions beyond conversation. Here's how to add a search tool:
from stella_sdk import tool
@tool
async def search_database(query: str, limit: int = 5) -> dict:
"""Search the database for relevant information.
Args:
query: The search query
limit: Maximum number of results
Returns:
Search results
"""
# Your database search logic here
results = await db.search(query, limit=limit)
return {
"found": len(results) > 0,
"results": [
{"title": r.title, "content": r.content}
for r in results
]
}
@tool
async def create_task(title: str, description: str) -> dict:
"""Create a new task in the task management system.
Args:
title: Task title
description: Task description
Returns:
Created task details
"""
task = await task_service.create(title=title, description=description)
return {
"task_id": task.id,
"status": "created",
"message": f"Task '{title}' created successfully"
}
Register tools in your agent:
from .tools import search_database, create_task
class MyCustomAgent(BaseAgent):
def __init__(self):
super().__init__()
# ... other initialization
# Register tools
self.register_tool(search_database)
self.register_tool(create_task)
Handling Tool Callsβ
When the LLM wants to use a tool, handle it in your response generation:
async def generate_response(self, user_input: str) -> str:
messages = [
{"role": "system", "content": self.system_prompt},
*self.history
]
response = await self.openai.chat.completions.create(
model=settings.openai_model,
messages=messages,
tools=self.get_tool_definitions()
)
# Handle tool calls
while response.choices[0].message.tool_calls:
tool_calls = response.choices[0].message.tool_calls
messages.append(response.choices[0].message)
for tool_call in tool_calls:
# Execute the tool
result = await self.execute_tool(
tool_call.function.name,
json.loads(tool_call.function.arguments)
)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})
# Get next response
response = await self.openai.chat.completions.create(
model=settings.openai_model,
messages=messages,
tools=self.get_tool_definitions()
)
return response.choices[0].message.content
Accessing Chat Historyβ
STELLA automatically records all messages exchanged during a sessionβyou don't need to implement any recording logic. Your agent can retrieve this conversation history using the built-in get_chat_history() method.
No setup required: The platform handles message recording, storage, and retrieval. Your agent just calls the method.
This is useful for:
- Building context for LLM prompts
- Resuming conversations after agent restart
- Analyzing conversation patterns
Basic Usageβ
class MyAgent(BaseAgent):
async def process(self, input: AgentInput) -> AsyncIterator[AgentOutput]:
# Get recent conversation history
history = await self.get_chat_history(limit=20)
# Build context for LLM
context = "\n".join([
f"{msg.role}: {msg.content}"
for msg in history
])
# Use in your prompt
response = await self.llm.chat([
{"role": "system", "content": f"Previous conversation:\n{context}"},
{"role": "user", "content": input.text}
])
yield AgentOutput.text_final(input.session_id, response)
Method Referenceβ
await self.get_chat_history(
include_debug: bool = False, # Include debug/processing messages
limit: int = 100, # Max messages (up to 500)
) -> List[ChatMessage]
ChatMessage Structureβ
Each message includes:
id- Unique message identifiertimestamp- ISO 8601 timestamprole- Sender role ('user', 'assistant', 'system')content- Extracted text contentmessage_type- Type ('user_text', 'transcript', 'agent_text')envelope- Full original message envelope
Checking Availabilityβ
if self.has_history:
history = await self.get_chat_history()
Building and Deployingβ
Dockerfileβ
FROM python:3.11-slim
WORKDIR /app
# Install system dependencies
RUN apt-get update && apt-get install -y \
ffmpeg \
&& rm -rf /var/lib/apt/lists/*
# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy source code
COPY src/ ./src/
ENV PYTHONPATH=/app
CMD ["python", "-m", "src.agent"]
Build and Pushβ
# Build the image
docker build -t my-custom-agent:latest .
# Test locally
docker run --env-file .env my-custom-agent:latest
# Push to registry
docker tag my-custom-agent:latest registry.example.com/my-custom-agent:latest
docker push registry.example.com/my-custom-agent:latest
Register with STELLAβ
Add your agent to the STELLA configuration so it can be deployed:
agents:
- name: my-custom-agent
image: registry.example.com/my-custom-agent:latest
resources:
requests:
cpu: "250m"
memory: "512Mi"
limits:
cpu: "1000m"
memory: "2Gi"
Run Deployment Testsβ
Before deploying an agent, validate the same surfaces STELLA depends on at build and runtime:
agent.yamlshould parse cleanly and satisfy the canonical manifest schema- the package should include a
Dockerfileor point to a published image - Python dependencies should include
stella-ai-agent-sdk - the agent should start successfully with representative config
For built-in agents in this repository, these checks are enforced automatically in GitHub Actions:
npm run validate:agentsvalidates built-in manifests, Dockerfile presence, slug consistency, and SDK compatibility- Python startup smoke tests install the SDK and agents, then verify they initialize with sample config
For custom agents, validation happens individually per package. When a custom agent is uploaded or built, STELLA validates that package's manifest, Dockerfile, and SDK dependency before deployment.
If you are developing your agent in this repository, run the built-in validation checks locally before opening a PR:
npm run validate:agents
Best Practicesβ
- Error Handling: Always wrap external API calls in try/catch blocks
- Logging: Add comprehensive logging for debugging production issues
- Timeouts: Set appropriate timeouts for tool executions
- Rate Limiting: Respect API rate limits for external services
- Testing: Write unit tests for tools and response generation
Next Stepsβ
- Add Custom UI - Customize the frontend
- Base Agent Reference - Full API documentation
- Tools Reference - Advanced tool patterns