Personal digital assistants used to mean a Palm Pilot in your pocket. Today, a real AI personal assistant means something far more capable: a system that answers calls, manages schedules, searches knowledge bases, reads emails, and handles other daily operations for you.
This post goes over our example of that system: an open-source voice AI agent built on the SignalWire Agents SDK, designed for developers who want to deploy a fully functional, configurable personal assistant AI without building every integration from scratch.
Introducing Ethan: An AI Personal Assistant App
Ethan is a Python-based AI voice assistant that handles inbound phone calls, call routing, appointment booking, and other tasks. It's not a chatbot or a phone tree, but an AI voice assistant that conducts real conversations, executes real actions, and integrates with the tools your users already rely on.
This is a personal assistant AI that runs on a phone number. So when someone calls, Ethan answers, identifies the caller, understands their intent, and takes action.
Stack:
SignalWire Agents SDK: voice AI infrastructure and SWAIG function execution
FastAPI: web server and admin API
Google APIs: Calendar, Gmail, and Contacts via OAuth2
SQLAlchemy: database layer (SQLite default, PostgreSQL supported)
RAG knowledge base: SQLite or pgvector backend for document search
View the repo on GitHub to to get everything you need to build your own.
AI appointment scheduling, email management, and more: What Ethan handles
Customer mode vs. owner mode
This example supports two distinct user flows depending on who's calling, making decisions based on the split between customer-facing and owner-facing behavior.
Customer mode: What any inbound caller experiences.
Book, check, or cancel an appointment
Get business hours, location, services, pricing
Search FAQs and knowledge base
Leave a message or send an email
Transfer to the owner
Owner mode: Activated when the owner calls from their registered number.
Status briefing: unread emails, pending messages, upcoming appointments
Full inbox management via Gmail
Contact lookup by name or phone
Message review and deletion
Calendar management including invite responses
This pattern is directly applicable to any use case where a single AI assistant needs to behave differently depending on who's interacting with it.
def get_credentials(self, user_id: str = None) -> Optional[Credentials]:
"""Get stored credentials for a user, refreshing if needed"""
if not user_id:
return None
db = SessionLocal()
try:
token = UserGoogleToken.get_by_user(db, user_id)
if not token or not token.access_token:
return None
creds = Credentials(
token=token.access_token,
refresh_token=token.refresh_token,
token_uri=token.token_uri or "https://oauth2.googleapis.com/token",
client_id=self.client_id,
client_secret=self.client_secret,
scopes=token.scopes or SCOPES,
)
if token.expiry:
creds.expiry = token.expiry
if creds.expired and creds.refresh_token:
try:
creds.refresh(Request())
self._save_credentials(user_id, creds)
except Exception as e:
print(f"[GoogleAuth] Error refreshing credentials for user {user_id}: {e}")
return None
return creds
finally:
db.close()
def get_credentials(self, user_id: str = None) -> Optional[Credentials]:
"""Get stored credentials for a user, refreshing if needed"""
if not user_id:
return None
db = SessionLocal()
try:
token = UserGoogleToken.get_by_user(db, user_id)
if not token or not token.access_token:
return None
creds = Credentials(
token=token.access_token,
refresh_token=token.refresh_token,
token_uri=token.token_uri or "https://oauth2.googleapis.com/token",
client_id=self.client_id,
client_secret=self.client_secret,
scopes=token.scopes or SCOPES,
)
if token.expiry:
creds.expiry = token.expiry
if creds.expired and creds.refresh_token:
try:
creds.refresh(Request())
self._save_credentials(user_id, creds)
except Exception as e:
print(f"[GoogleAuth] Error refreshing credentials for user {user_id}: {e}")
return None
return creds
finally:
db.close()
def get_credentials(self, user_id: str = None) -> Optional[Credentials]:
"""Get stored credentials for a user, refreshing if needed"""
if not user_id:
return None
db = SessionLocal()
try:
token = UserGoogleToken.get_by_user(db, user_id)
if not token or not token.access_token:
return None
creds = Credentials(
token=token.access_token,
refresh_token=token.refresh_token,
token_uri=token.token_uri or "https://oauth2.googleapis.com/token",
client_id=self.client_id,
client_secret=self.client_secret,
scopes=token.scopes or SCOPES,
)
if token.expiry:
creds.expiry = token.expiry
if creds.expired and creds.refresh_token:
try:
creds.refresh(Request())
self._save_credentials(user_id, creds)
except Exception as e:
print(f"[GoogleAuth] Error refreshing credentials for user {user_id}: {e}")
return None
return creds
finally:
db.close()
def get_credentials(self, user_id: str = None) -> Optional[Credentials]:
"""Get stored credentials for a user, refreshing if needed"""
if not user_id:
return None
db = SessionLocal()
try:
token = UserGoogleToken.get_by_user(db, user_id)
if not token or not token.access_token:
return None
creds = Credentials(
token=token.access_token,
refresh_token=token.refresh_token,
token_uri=token.token_uri or "https://oauth2.googleapis.com/token",
client_id=self.client_id,
client_secret=self.client_secret,
scopes=token.scopes or SCOPES,
)
if token.expiry:
creds.expiry = token.expiry
if creds.expired and creds.refresh_token:
try:
creds.refresh(Request())
self._save_credentials(user_id, creds)
except Exception as e:
print(f"[GoogleAuth] Error refreshing credentials for user {user_id}: {e}")
return None
return creds
finally:
db.close()
AI appointment scheduling
Ethan connects directly to Google Calendar to handle the full appointment lifecycle over the phone. Callers can book a new appointment, check an existing one, or cancel, and the calendar updates in real time without a scheduling widget or any back-and-forth email thread.
This makes Ethan a practical foundation for any AI appointment scheduling solution where the interaction happens over voice rather than a web form.
Available SignalWire AI Gateway (SWAIG) functions:
check_calendar_availability: query open slots by date
book_appointment: create a confirmed calendar event
cancel_appointment: remove an existing booking
update_appointment: modify time or details (owner mode)
respond_to_invite: RSVP to calendar invites (owner mode)
def book_appointment(
self,
start_time: str,
end_time: str,
appointment_type: str,
attendee_name: str,
attendee_email: str,
attendee_phone: str = "",
notes: str = "",
calendar_id: str = None,
user_id: str = None
) -> Dict
def book_appointment(
self,
start_time: str,
end_time: str,
appointment_type: str,
attendee_name: str,
attendee_email: str,
attendee_phone: str = "",
notes: str = "",
calendar_id: str = None,
user_id: str = None
) -> Dict
def book_appointment(
self,
start_time: str,
end_time: str,
appointment_type: str,
attendee_name: str,
attendee_email: str,
attendee_phone: str = "",
notes: str = "",
calendar_id: str = None,
user_id: str = None
) -> Dict
def book_appointment(
self,
start_time: str,
end_time: str,
appointment_type: str,
attendee_name: str,
attendee_email: str,
attendee_phone: str = "",
notes: str = "",
calendar_id: str = None,
user_id: str = None
) -> Dict
AI email management
Ethan reads, archives, deletes, and responds to emails via Gmail all through voice. For the business owner calling in from their registered number, Ethan switches into a privileged mode and can give a full inbox briefing on demand.
Functions include:
send_email: Send email
email_owner: Email the business owner
Which allow callers to contact the business owner without having to track down an email address.
Meanwhile, functions like
delete_email: Delete email (owner only)
mark_email_read: Mark as read (owner only)
archive_email: Archive email (owner only)
Allow the owner to manage email through Ethan.
def send_message_notification(
self,
caller_name: str,
caller_phone: str,
message: str,
urgency: str = "normal",
user_id: str = None
) -> Dict:
"""Send notification to owner about a new message"""
info = self._get_business_info(user_id)
owner_email = info.get("owner_email")
if not owner_email:
return {
"success": False,
"error": "Owner email not configured"
}
urgency_prefix = "URGENT: " if urgency == "urgent" else ""
subject = f"{urgency_prefix}New Message from {caller_name}"
def send_message_notification(
self,
caller_name: str,
caller_phone: str,
message: str,
urgency: str = "normal",
user_id: str = None
) -> Dict:
"""Send notification to owner about a new message"""
info = self._get_business_info(user_id)
owner_email = info.get("owner_email")
if not owner_email:
return {
"success": False,
"error": "Owner email not configured"
}
urgency_prefix = "URGENT: " if urgency == "urgent" else ""
subject = f"{urgency_prefix}New Message from {caller_name}"
def send_message_notification(
self,
caller_name: str,
caller_phone: str,
message: str,
urgency: str = "normal",
user_id: str = None
) -> Dict:
"""Send notification to owner about a new message"""
info = self._get_business_info(user_id)
owner_email = info.get("owner_email")
if not owner_email:
return {
"success": False,
"error": "Owner email not configured"
}
urgency_prefix = "URGENT: " if urgency == "urgent" else ""
subject = f"{urgency_prefix}New Message from {caller_name}"
def send_message_notification(
self,
caller_name: str,
caller_phone: str,
message: str,
urgency: str = "normal",
user_id: str = None
) -> Dict:
"""Send notification to owner about a new message"""
info = self._get_business_info(user_id)
owner_email = info.get("owner_email")
if not owner_email:
return {
"success": False,
"error": "Owner email not configured"
}
urgency_prefix = "URGENT: " if urgency == "urgent" else ""
subject = f"{urgency_prefix}New Message from {caller_name}"
Inbound call handling
Ethan is the first point of contact for every incoming call. It identifies returning callers via Google Contacts, routes owner vs. customer calls into separate flows, and transfers to a live person when needed. It can also take and log messages when the owner is unavailable.
Knowledge base & FAQ search
Upload documents to Ethan's RAG knowledge base powered by the SignalWire Datasphere API. When a caller asks a question the FAQ doesn't cover, Ethan runs a similarity search across indexed documents and synthesizes a response. This is what separates a real AI personal assistant app from a basic phone menu.
def search(self, query: str, count: int = 5, user_id: str = None) -> List[Dict]:
"""Search the knowledge base for a user (for testing/admin use)"""
index_path = self._get_user_index_path(user_id)
if not index_path.exists():
return []
try:
from signalwire_agents.search import SearchEngine
from signalwire_agents.search.query_processor import preprocess_query
engine = SearchEngine(
backend="sqlite",
index_path=str(index_path),
)
processed = preprocess_query(
query,
language="en",
vector=True,
model_name="sentence-transformers/all-MiniLM-L6-v2",
)
results = engine.search(
query_vector=processed.get("vector", []),
enhanced_text=processed.get("enhanced_text", query),
count=count,
original_query=query,
)
return [
{
"content": r.get("content", ""),
"score": r.get("score", 0),
"metadata": r.get("metadata", {}),
}
for r in results
]
def search(self, query: str, count: int = 5, user_id: str = None) -> List[Dict]:
"""Search the knowledge base for a user (for testing/admin use)"""
index_path = self._get_user_index_path(user_id)
if not index_path.exists():
return []
try:
from signalwire_agents.search import SearchEngine
from signalwire_agents.search.query_processor import preprocess_query
engine = SearchEngine(
backend="sqlite",
index_path=str(index_path),
)
processed = preprocess_query(
query,
language="en",
vector=True,
model_name="sentence-transformers/all-MiniLM-L6-v2",
)
results = engine.search(
query_vector=processed.get("vector", []),
enhanced_text=processed.get("enhanced_text", query),
count=count,
original_query=query,
)
return [
{
"content": r.get("content", ""),
"score": r.get("score", 0),
"metadata": r.get("metadata", {}),
}
for r in results
]
def search(self, query: str, count: int = 5, user_id: str = None) -> List[Dict]:
"""Search the knowledge base for a user (for testing/admin use)"""
index_path = self._get_user_index_path(user_id)
if not index_path.exists():
return []
try:
from signalwire_agents.search import SearchEngine
from signalwire_agents.search.query_processor import preprocess_query
engine = SearchEngine(
backend="sqlite",
index_path=str(index_path),
)
processed = preprocess_query(
query,
language="en",
vector=True,
model_name="sentence-transformers/all-MiniLM-L6-v2",
)
results = engine.search(
query_vector=processed.get("vector", []),
enhanced_text=processed.get("enhanced_text", query),
count=count,
original_query=query,
)
return [
{
"content": r.get("content", ""),
"score": r.get("score", 0),
"metadata": r.get("metadata", {}),
}
for r in results
]
def search(self, query: str, count: int = 5, user_id: str = None) -> List[Dict]:
"""Search the knowledge base for a user (for testing/admin use)"""
index_path = self._get_user_index_path(user_id)
if not index_path.exists():
return []
try:
from signalwire_agents.search import SearchEngine
from signalwire_agents.search.query_processor import preprocess_query
engine = SearchEngine(
backend="sqlite",
index_path=str(index_path),
)
processed = preprocess_query(
query,
language="en",
vector=True,
model_name="sentence-transformers/all-MiniLM-L6-v2",
)
results = engine.search(
query_vector=processed.get("vector", []),
enhanced_text=processed.get("enhanced_text", query),
count=count,
original_query=query,
)
return [
{
"content": r.get("content", ""),
"score": r.get("score", 0),
"metadata": r.get("metadata", {}),
}
for r in results
]
Multi-tenant support
Ethan supports multiple businesses on separate phone numbers from a single deployment. Each tenant has its own configuration, business hours, FAQs, Google integration, and knowledge base, all managed through a web dashboard.
Get started with your own AI personal assistant app
Ethan is open source and built to be extended. The core agent is in agent.py, integrations live in the services/ directory, and the admin panel is a FastAPI app with Jinja2 templates.
Clone the repo, configure your .env, point a SignalWire phone number at the webhook, and Ethan is live.
Full setup documentation, including Google OAuth configuration, SignalWire phone number setup, and deployment options for Docker, Heroku, and cloud VMs, is in the GitHub repo. As you’re building, chat with fellow developers and get assistance from our team of experts in our community Discord.
