Overview
Use the Python MCP SDK to build custom AI agents that can schedule meetings, find availability, and manage calendars through the Syncline MCP server.Prerequisites
1
Install Syncline MCP Server
Copy
npm install -g @kekwanulabs/syncline-mcp-server
2
Install Python MCP SDK
Copy
pip install mcp
3
Get Platform API Key
Sign up at syncline.run/developer/login
Quick Start
Create a simple agent that finds meeting availability:Copy
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
# Configure Syncline MCP server
server_params = StdioServerParameters(
command="npx",
args=["-y", "@kekwanulabs/syncline-mcp-server"],
env={"SYNCLINE_API_KEY": "sk_live_your_api_key_here"}
)
# Connect to server
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# Initialize connection
await session.initialize()
# List available tools
tools = await session.list_tools()
print(f"Available tools: {[tool.name for tool in tools]}")
# Find mutual availability
result = await session.call_tool(
"find_mutual_availability",
{
"attendees": ["alice@example.com", "bob@example.com"],
"duration_minutes": 30
}
)
print(f"Available time slots:\n{result.content[0].text}")
if __name__ == "__main__":
asyncio.run(main())
Copy
python agent.py
Copy
Available tools: ['find_mutual_availability', 'schedule_meeting', 'check_availability', 'update_preferences']
Available time slots:
{
"slots": [
{
"start_time": "2025-01-22T14:00:00-08:00",
"end_time": "2025-01-22T14:30:00-08:00",
"score": 0.95,
"quality": "excellent"
},
...
]
}
Building a Meeting Scheduler Agent
Create an AI agent that can schedule meetings via natural language:Copy
import asyncio
import json
from datetime import datetime, timedelta
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
class MeetingSchedulerAgent:
def __init__(self, api_key):
self.api_key = api_key
self.session = None
async def __aenter__(self):
server_params = StdioServerParameters(
command="npx",
args=["-y", "@kekwanulabs/syncline-mcp-server"],
env={"SYNCLINE_API_KEY": self.api_key}
)
self.transport = await stdio_client(server_params).__aenter__()
read, write = self.transport
self.session = await ClientSession(read, write).__aenter__()
await self.session.initialize()
return self
async def __aexit__(self, *args):
await self.session.__aexit__(*args)
await self.transport[0].__aexit__(*args)
async def find_availability(self, attendees, duration=30):
"""Find time slots where all attendees are available."""
result = await self.session.call_tool(
"find_mutual_availability",
{
"attendees": attendees,
"duration_minutes": duration
}
)
return json.loads(result.content[0].text)
async def schedule_meeting(self, attendees, start_time, title, duration=30):
"""Schedule a meeting with Google Meet link."""
result = await self.session.call_tool(
"schedule_meeting",
{
"attendees": attendees,
"start_time": start_time,
"title": title,
"duration_minutes": duration
}
)
return json.loads(result.content[0].text)
async def check_calendar(self, email, days_ahead=7):
"""Check a user's calendar for the next N days."""
end_date = datetime.now() + timedelta(days=days_ahead)
result = await self.session.call_tool(
"check_availability",
{
"email": email,
"start_date": datetime.now().isoformat(),
"end_date": end_date.isoformat()
}
)
return json.loads(result.content[0].text)
async def update_preferences(self, email, preferences):
"""Update user scheduling preferences."""
result = await self.session.call_tool(
"update_preferences",
{
"email": email,
"preferences": preferences
}
)
return json.loads(result.content[0].text)
async def main():
# Create agent
async with MeetingSchedulerAgent("sk_live_your_api_key_here") as agent:
# Find when Alice and Bob are both free
print("Finding availability...")
availability = await agent.find_availability(
["alice@example.com", "bob@example.com"],
duration=30
)
# Get the best time slot
best_slot = availability["slots"][0]
print(f"\nBest time slot: {best_slot['start_time']}")
print(f"Quality score: {best_slot['score']}")
# Schedule the meeting
print("\nScheduling meeting...")
meeting = await agent.schedule_meeting(
attendees=["alice@example.com", "bob@example.com"],
start_time=best_slot["start_time"],
title="Product Demo",
duration=30
)
print(f"\n✓ Meeting scheduled!")
print(f" Meeting ID: {meeting['meeting_id']}")
print(f" Google Meet: {meeting['google_meet_link']}")
print(f" Calendar event: {meeting['calendar_event_link']}")
if __name__ == "__main__":
asyncio.run(main())
Advanced Examples
Conversational Scheduling Bot
Build a bot that schedules meetings through natural language:Copy
import asyncio
import re
from datetime import datetime, timedelta
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
class SchedulingBot:
def __init__(self, api_key):
self.api_key = api_key
self.session = None
async def __aenter__(self):
server_params = StdioServerParameters(
command="npx",
args=["-y", "@kekwanulabs/syncline-mcp-server"],
env={"SYNCLINE_API_KEY": self.api_key}
)
self.transport = await stdio_client(server_params).__aenter__()
read, write = self.transport
self.session = await ClientSession(read, write).__aenter__()
await self.session.initialize()
return self
async def __aexit__(self, *args):
if self.session:
await self.session.__aexit__(*args)
if hasattr(self, 'transport'):
await self.transport[0].__aexit__(*args)
def parse_intent(self, message):
"""Parse user intent from natural language."""
message = message.lower()
# Extract emails
emails = re.findall(r'[\w\.-]+@[\w\.-]+\.\w+', message)
# Detect intent
if "schedule" in message or "book" in message:
return {"intent": "schedule", "attendees": emails}
elif "available" in message or "free" in message:
return {"intent": "find_availability", "attendees": emails}
elif "check" in message and "calendar" in message:
return {"intent": "check_calendar", "email": emails[0] if emails else None}
else:
return {"intent": "unknown"}
async def handle_message(self, message):
"""Handle a natural language scheduling request."""
intent_data = self.parse_intent(message)
if intent_data["intent"] == "find_availability":
availability = await self.session.call_tool(
"find_mutual_availability",
{
"attendees": intent_data["attendees"],
"duration_minutes": 30
}
)
return self.format_availability_response(availability)
elif intent_data["intent"] == "schedule":
# First find availability
availability = await self.session.call_tool(
"find_mutual_availability",
{
"attendees": intent_data["attendees"],
"duration_minutes": 30
}
)
# Parse response and get best slot
slots = json.loads(availability.content[0].text)["slots"]
best_slot = slots[0]
# Schedule the meeting
meeting = await self.session.call_tool(
"schedule_meeting",
{
"attendees": intent_data["attendees"],
"start_time": best_slot["start_time"],
"title": "Meeting",
"duration_minutes": 30
}
)
return self.format_meeting_response(meeting)
elif intent_data["intent"] == "check_calendar":
calendar = await self.session.call_tool(
"check_availability",
{
"email": intent_data["email"],
"start_date": datetime.now().isoformat(),
"end_date": (datetime.now() + timedelta(days=7)).isoformat()
}
)
return self.format_calendar_response(calendar)
else:
return "I can help you schedule meetings, find availability, or check calendars. Try asking: 'When are alice@example.com and bob@example.com free?'"
def format_availability_response(self, availability):
slots = json.loads(availability.content[0].text)["slots"]
response = "Here are the available time slots:\n\n"
for i, slot in enumerate(slots[:5], 1):
response += f"{i}. {slot['start_time']} (quality: {slot['quality']})\n"
return response
def format_meeting_response(self, meeting):
data = json.loads(meeting.content[0].text)
return f"✓ Meeting scheduled!\nGoogle Meet: {data['google_meet_link']}\nCalendar: {data['calendar_event_link']}"
def format_calendar_response(self, calendar):
data = json.loads(calendar.content[0].text)
return f"Calendar checked. Free slots: {len(data.get('free_slots', []))}, Busy slots: {len(data.get('busy_slots', []))}"
async def chat_loop():
"""Interactive chat loop for scheduling."""
async with SchedulingBot("sk_live_your_api_key_here") as bot:
print("Scheduling Bot Ready! Ask me to schedule meetings or find availability.")
print("Type 'quit' to exit.\n")
while True:
user_input = input("You: ")
if user_input.lower() in ["quit", "exit", "bye"]:
print("Goodbye!")
break
response = await bot.handle_message(user_input)
print(f"Bot: {response}\n")
if __name__ == "__main__":
asyncio.run(chat_loop())
Copy
You: When are alice@example.com and bob@example.com free?
Bot: Here are the available time slots:
1. 2025-01-22T14:00:00-08:00 (quality: excellent)
2. 2025-01-22T15:00:00-08:00 (quality: good)
3. 2025-01-23T10:00:00-08:00 (quality: good)
4. 2025-01-23T14:00:00-08:00 (quality: excellent)
5. 2025-01-24T09:00:00-08:00 (quality: fair)
You: Schedule a meeting with alice@example.com and bob@example.com
Bot: ✓ Meeting scheduled!
Google Meet: https://meet.google.com/abc-defg-hij
Calendar: https://calendar.google.com/calendar/event?eid=...
Error Handling
Handle errors gracefully:Copy
import asyncio
from mcp.types import McpError
async def safe_schedule_meeting(agent, attendees, start_time, title):
"""Schedule meeting with error handling."""
try:
meeting = await agent.schedule_meeting(
attendees=attendees,
start_time=start_time,
title=title
)
return {"success": True, "meeting": meeting}
except McpError as e:
print(f"MCP Error: {e}")
return {"success": False, "error": str(e)}
except Exception as e:
print(f"Unexpected error: {e}")
return {"success": False, "error": "An unexpected error occurred"}
# Usage
async with MeetingSchedulerAgent("sk_live_your_key") as agent:
result = await safe_schedule_meeting(
agent,
["alice@example.com", "bob@example.com"],
"2025-01-22T14:00:00-08:00",
"Product Demo"
)
if result["success"]:
print(f"✓ Meeting scheduled: {result['meeting']['google_meet_link']}")
else:
print(f"✗ Failed to schedule: {result['error']}")
Best Practices
Use Environment Variables
Store API keys securely:Copy
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("SYNCLINE_API_KEY")
if not api_key:
raise ValueError("SYNCLINE_API_KEY environment variable not set")
async with MeetingSchedulerAgent(api_key) as agent:
# Use agent...
Implement Retry Logic
Handle transient failures:Copy
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def schedule_with_retry(agent, attendees, start_time, title):
"""Schedule meeting with automatic retries."""
return await agent.schedule_meeting(attendees, start_time, title)
# Usage
try:
meeting = await schedule_with_retry(agent, attendees, start_time, title)
print(f"✓ Scheduled: {meeting['meeting_id']}")
except Exception as e:
print(f"✗ Failed after 3 attempts: {e}")
Validate Input
Validate user input before calling tools:Copy
import re
from email_validator import validate_email, EmailNotValidError
def validate_attendees(emails):
"""Validate list of email addresses."""
validated = []
for email in emails:
try:
v = validate_email(email)
validated.append(v.email)
except EmailNotValidError as e:
print(f"Invalid email {email}: {e}")
return validated
# Usage
user_emails = ["alice@example.com", "not-an-email", "bob@example.com"]
valid_emails = validate_attendees(user_emails)
# Returns: ["alice@example.com", "bob@example.com"]
Testing
Write tests for your agent:Copy
import pytest
import asyncio
from unittest.mock import Mock, patch
@pytest.mark.asyncio
async def test_find_availability():
"""Test finding availability."""
# Mock the MCP session
mock_session = Mock()
mock_session.call_tool.return_value = Mock(
content=[Mock(text='{"slots": [{"start_time": "2025-01-22T14:00:00Z", "score": 0.95}]}')]
)
agent = MeetingSchedulerAgent("test_key")
agent.session = mock_session
# Test
result = await agent.find_availability(
["alice@example.com", "bob@example.com"],
duration=30
)
# Assert
assert len(result["slots"]) > 0
assert result["slots"][0]["score"] == 0.95