""" title: Google Calendar Create Event Pipe author: Your Name version: 0.1.0 license: MIT """ from pydantic import BaseModel, Field from typing import Union, Generator, Iterator import requests class Pipe: class Valves(BaseModel): # API key for the Composio API (used for the x-api-key header) COMPOSIO_API_KEY: str = Field( default="", description="API key for accessing Composio services.", ) # Endpoint URL for creating a Google Calendar event GOOGLECALENDAR_API_URL: str = Field( default="https://backend.composio.dev/api/v2/actions/GOOGLECALENDAR_CREATE_EVENT/execute", description="Endpoint URL for the Google Calendar create event action.", ) def __init__(self): self.type = "external" self.valves = self.Valves() def pipe(self, body: dict, __user__: dict) -> Union[str, Generator, Iterator]: """ Create a Google Calendar event using the Composio API. Expected parameters in `body`: - start_datetime: (str) ISO8601 format datetime (e.g., "2025-03-16T13:00:00") - event_duration_hour: (optional, default "1") Duration hours as string or int - event_duration_minutes: (optional, default "0") Duration minutes as string or int - timezone: (optional, default "America/New_York") Timezone string - connectedAccountId: (str) The ID of the connected Google account in Composio Returns the JSON response from the API or an error message. """ print("Pipe: Google Calendar Create Event") # Check that the API key is provided via the valves configuration. if not self.valves.COMPOSIO_API_KEY: raise Exception("COMPOSIO_API_KEY not provided in the valves.") # Extract parameters from the body, providing defaults if necessary. start_datetime = body.get("start_datetime", "2025-03-16T13:00:00") event_duration_hour = str(body.get("event_duration_hour", "1")) event_duration_minutes = str(body.get("event_duration_minutes", "0")) timezone = body.get("timezone", "America/New_York") connected_account_id = body.get("connectedAccountId", "") if not connected_account_id: raise Exception("connectedAccountId must be provided in the body.") # Prepare headers and payload as required by the API. headers = { "Content-Type": "application/json", "x-api-key": self.valves.COMPOSIO_API_KEY, } payload = { "authConfig": { "parameters": [] }, "appName": "googlecalendar", "entityId": "", "input": { "start_datetime": start_datetime, "event_duration_hour": event_duration_hour, "event_duration_minutes": event_duration_minutes, "timezone": timezone }, "connectedAccountId": connected_account_id } print("Payload:", payload) try: # Call the Composio API endpoint to create the calendar event. response = requests.post( url=self.valves.GOOGLECALENDAR_API_URL, headers=headers, json=payload, ) response.raise_for_status() return response.json() except Exception as e: return f"Error: {e}"