Skip to main content

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

npm install -g @kekwanulabs/syncline-mcp-server
2

Install Python MCP SDK

pip install mcp
3

Get Platform API Key

Quick Start

Create a simple agent that finds meeting availability:
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())
Run the script:
python agent.py
Output:
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:
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:
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())
Example conversation:
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:
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:
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:
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:
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:
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