Spaces:
Running
Running
File size: 4,360 Bytes
a66d4bd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """
Google Calendar Agent – specialized agent with only Calendar MCP tools.
Uses MCPServerStdio (local subprocess) with create_static_tool_filter so the
agent sees *only* calendar / event tools. Zero network overhead.
"""
import asyncio
import logging
from openai import AsyncOpenAI
from agents import Agent, Runner, OpenAIChatCompletionsModel
from agents.model_settings import ModelSettings
try:
from .google_mcp_config import (
LONGCAT_API_KEY, LONGCAT_BASE_URL, MODEL_NAME,
CALENDAR_TOOLS, create_google_mcp_server, USER_GOOGLE_EMAIL,
)
except ImportError:
from google_mcp_config import (
LONGCAT_API_KEY, LONGCAT_BASE_URL, MODEL_NAME,
CALENDAR_TOOLS, create_google_mcp_server, USER_GOOGLE_EMAIL,
)
logger = logging.getLogger(__name__)
SYSTEM_PROMPT = """\
You are a specialized Google Calendar assistant. You can list calendars,
query events, create / modify / delete events, and check free-busy
availability using the available tools.
Capabilities:
- **List calendars** the user has access to
- **Get events** in a date range (supports filtering by calendar)
- **Create events** with title, description, start/end times, attendees,
location, recurrence rules, and reminders
- **Modify events** — change any field on an existing event
- **Delete events** by event ID
- **Free/busy queries** — check availability for one or more calendars
in a given time range
Rules:
1. The user's Google email is provided in the query — use it for every
tool call in the `user_google_email` parameter. NEVER ask the user
for their email; it is always supplied.
2. Use ISO 8601 format for dates/times (e.g. "2025-01-15T09:00:00-05:00").
If the user gives a natural-language date like "next Tuesday at 3 PM",
convert it to ISO 8601 before calling the tool.
3. When creating or modifying events involving other attendees, list
their email addresses explicitly.
4. After any create/modify/delete, confirm the change with the event
title, date, and time.
5. For recurring events, use RRULE syntax (e.g. "RRULE:FREQ=WEEKLY;COUNT=10").
"""
class GoogleCalendarAgent:
"""Thin wrapper around the OpenAI Agent SDK wired to Google Calendar tools."""
def __init__(self, model: str = MODEL_NAME):
self.model = model
self._client = AsyncOpenAI(
api_key=LONGCAT_API_KEY,
base_url=LONGCAT_BASE_URL,
timeout=30.0,
)
# ── factory helpers ──────────────────────────────────────────────────
def _create_mcp_server(self):
"""Spawn a local MCP subprocess with only Calendar tools loaded."""
return create_google_mcp_server(service="calendar", tool_names=CALENDAR_TOOLS)
def _create_agent(self, mcp_server) -> Agent:
return Agent(
name="Google Calendar Agent",
instructions=SYSTEM_PROMPT,
mcp_servers=[mcp_server],
model=OpenAIChatCompletionsModel(
model=self.model,
openai_client=self._client,
),
model_settings=ModelSettings(tool_choice="auto"),
)
# ── public API ───────────────────────────────────────────────────────
async def run(self, query: str) -> str:
"""Spawn MCP connection, run a single query, then clean up."""
mcp_server = self._create_mcp_server()
async with mcp_server:
agent = self._create_agent(mcp_server)
logger.info("Google Calendar MCP connected – agent ready")
result = await Runner.run(agent, input=query)
return result.final_output
# ─── CLI entry point ──────────────────────────────────────────────────────────
async def main():
agent = GoogleCalendarAgent()
resp = await agent.run(
"Show my events for this week. My email is user@example.com"
)
print("Agent Response:\n", resp)
if __name__ == "__main__":
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
asyncio.run(main())
|