To make voice agents useful, they often need deep integrations with third-party services by way of function calls (tool calls). In this post, we'll walk through how to create a voice agent that can create, read, update and delete events on your Google Calendar.
If you'd like to skip ahead and just get the code, you can find it here.
For server-side usage of your Google Calendar OAuth2.0 credentials, you'll need to use a refresh token.
Refresh tokens are long-lived credentials that can be used to access resources on behalf of a user. They are issued to apps, not individuals, and can be used to obtain new access tokens.
In this pattern, you'll need to authenticate with Google once to get the refresh token. You can then use the refresh token to obtain new access tokens as needed. To do this, you'll need to make a simple local app so that you can specify the redirect URI as http://localhost:3000.
The pattern is as follows:
1.
2.
3.
4.
5.
Creating the app is beyond the scope of this post, but you can just use Vite or Next.js to create a simple app. Ideally, you will have a redirect page that gets the authorization code from the query params and then sends it to a backend service (or API route if you are using Next.js) that will exchange the code for a refresh token. You can store this refresh token safely in your database for use later on.
Once you have set up your Google Calendar OAuth2.0 credentials, you can create a voice agent that can create, read, update and delete events on your Google Calendar.
The voice agent will need to be able to use the Google Calendar API.
Start a new Pipecat project and install the following packages:
poetry add google-api-python-client google-auth-httplib2 google-auth-oauthlib
Use the following code or similar to register four different functions that are available to your agent:
create_calendar_eventget_free_availabilityupdate_calendar_eventcancel_calendar_event tools = [
ChatCompletionToolParam(
type="function",
function={
"name": "create_calendar_event",
"description": "Create a new calendar event",
"parameters": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Title of the event given the context of the conversation"
},
"description": {
"type": "string",
"description": "Description of the event given the context of the conversation"
},
"start_time": {
"type": "string",
"description": "Start time in ISO format (e.g., 2024-03-20T14:00:00) in the user's timezone"
},
"attendees": {
"type": "array",
"items": {"type": "string"},
"description": "List of attendee email addresses"
}
},
"required": ["title", "start_time"],
},
},
),
ChatCompletionToolParam(
type="function",
function={
"name": "get_free_availability",
"description": "Get free time slots in the calendar for a specified time range",
"parameters": {
"type": "object",
"properties": {
"start_time": {
"type": "string",
"description": "Start time in ISO format (e.g., 2024-03-20T09:00:00) in the user's timezone"
},
"end_time": {
"type": "string",
"description": "End time in ISO format (e.g., 2024-03-20T17:00:00) in the user's timezone"
}
},
"required": ["start_time", "end_time"],
},
},
),
ChatCompletionToolParam(
type="function",
function={
"name": "update_calendar_event",
"description": "Update an existing calendar event",
"parameters": {
"type": "object",
"properties": {
"event_id": {
"type": "string",
"description": "ID of the event to update"
},
"title": {
"type": "string",
"description": "New title of the event (optional)"
},
"description": {
"type": "string",
"description": "New description of the event (optional)"
},
"start_time": {
"type": "string",
"description": "New start time in ISO format (e.g., 2024-03-20T14:00:00) in the user's timezone"
}
},
"required": ["event_id"],
},
},
),
ChatCompletionToolParam(
type="function",
function={
"name": "cancel_calendar_event",
"description": "Cancel/delete an existing calendar event",
"parameters": {
"type": "object",
"properties": {
"event_id": {
"type": "string",
"description": "ID of the event to cancel"
}
},
"required": ["event_id"],
},
},
),
]
Ensure the functions are registered with the LLM.
llm.register_function( "create_calendar_event", calendar_service.handle_create_calendar_event, start_callback=calendar_service.start_create_calendar_event ) llm.register_function( "get_free_availability", calendar_service.handle_get_free_availability, start_callback=calendar_service.start_get_free_availability ) llm.register_function( "update_calendar_event", calendar_service.handle_update_calendar_event, start_callback=calendar_service.start_update_calendar_event ) llm.register_function( "cancel_calendar_event", calendar_service.handle_cancel_calendar_event, start_callback=calendar_service.start_cancel_calendar_event )
You will also need to create the functions that will be called when the agent uses the tool. I will only list the get_free_availability function as an example. You can find the others in the Github repo.
async def get_free_availability(self, start_time_str, end_time_str):
try:
service = self.get_calendar_service()
start_time = datetime.fromisoformat(start_time_str).replace(tzinfo=self.aedt)
end_time = datetime.fromisoformat(end_time_str).replace(tzinfo=self.aedt)
start_time_utc = start_time.astimezone(timezone.utc)
end_time_utc = end_time.astimezone(timezone.utc)
body = {
'timeMin': start_time_utc.isoformat(),
'timeMax': end_time_utc.isoformat(),
'items': [{'id': 'primary'}]
}
freebusy_response = await asyncio.get_event_loop().run_in_executor(
None,
lambda: service.freebusy().query(body=body).execute()
)
free_slots = self._process_freebusy_response(freebusy_response, start_time, end_time)
if free_slots:
free_slots_text = "\n".join([
f"- Available from {slot['start'].strftime('%I:%M %p')} to {slot['end'].strftime('%I:%M %p')}"
for slot in free_slots
])
return {
"available_slots": free_slots_text,
"count": len(free_slots)
}
else:
return {
"available_slots": "No free time slots found in the specified range",
"count": 0
}
except Exception as e:
logger.error(f"Error getting free availability: {str(e)}")
return {
"available_slots": f"Sorry, I encountered an error: {str(e)}",
"count": 0
}
Register your system prompt and first message in the messages variable. Typically, the main system prompt is always available at the top of the context window, while you might summarize, then concatenate, the messages in conversation history. This is to ensure that the context window doesn't grow indefinitely. As an example, this is what I'm using in this example.
current_time = datetime.now() name = "Matthew" email = "matthew@gmail.com" location = "Melbourne, Australia" messages = [ { "role": "system", "content": f"""You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way. Instructions: 1. If the user asks for availability, always use the get_free_availability function. When you respond to them, only give them times between 9am and 5pm where there is availability. 2. If the user asks to create a calendar event, use the create_calendar_event function. Create it for 30 minutes. Do not add any other attendees. 3. If the user asks to update a calendar event, use the update_calendar_event function. 4. If the user asks to cancel a calendar event, use the cancel_calendar_event function. You don't need to mention these details, but so that you have them for your own reference. 1. The current date and time is {current_time.strftime('%B %d, %Y %I:%M %p')} - and their current timezone is AEDT (Sydney time). 2. The user's name is {name}. 3. The user's email is {email}. 4. The user is located in {location}. """, }, { "role": "system", "content": "Start by introducing yourself as a calendar assistant and ask the user what they would like to do." } ]
To test your voice agent, open up your calendar and open up your terminal side by side. Then run the following command to start the server:
poetry run python server.py
Click on the start link to open a Daily room and you should hear your voice agent introduce itself and get started.
You can ask it to do things like check your availability, create an event, update an event or cancel an event.
You can now deploy your voice agent to the cloud and use it in production.
To do so, build the Docker container, then create an ECS task that runs the container. You might also want to set a DNS endpoint for the ECS task so that you can call it from your website and begin speaking to your voice agent.
Integrating apps and APIs with voice agents is much easier than you'd think. You definitely don't need VAPI or Retell, so don't let their limitations hold you back.
Happy coding!