diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/create_event.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/create_event.py index 37bcf083e..a8183314a 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/create_event.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/create_event.py @@ -168,20 +168,46 @@ def create_create_calendar_event_tool( f"Creating calendar event: summary='{final_summary}', connector={actual_connector_id}" ) + tz = context.get("timezone", "UTC") + if ( connector.connector_type == SearchSourceConnectorType.COMPOSIO_GOOGLE_CALENDAR_CONNECTOR ): - from app.utils.google_credentials import build_composio_credentials - cca_id = connector.config.get("composio_connected_account_id") - if cca_id: - creds = build_composio_credentials(cca_id) - else: + if not cca_id: return { "status": "error", "message": "Composio connected account ID not found for this connector.", } + + from app.services.composio_service import ComposioService + + ( + event_id, + html_link, + error, + ) = await ComposioService().create_calendar_event( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + summary=final_summary, + start_datetime=final_start_datetime, + end_datetime=final_end_datetime, + timezone=tz, + description=final_description, + location=final_location, + attendees=final_attendees, + ) + if error: + return {"status": "error", "message": error} + created = { + "id": event_id, + "summary": final_summary, + "htmlLink": html_link, + } + logger.info( + f"Calendar event created via Composio: id={event_id}, summary={final_summary}" + ) else: config_data = dict(connector.config) @@ -211,70 +237,69 @@ def create_create_calendar_event_tool( expiry=datetime.fromisoformat(exp) if exp else None, ) - service = await asyncio.get_event_loop().run_in_executor( - None, lambda: build("calendar", "v3", credentials=creds) - ) - - tz = context.get("timezone", "UTC") - event_body: dict[str, Any] = { - "summary": final_summary, - "start": {"dateTime": final_start_datetime, "timeZone": tz}, - "end": {"dateTime": final_end_datetime, "timeZone": tz}, - } - if final_description: - event_body["description"] = final_description - if final_location: - event_body["location"] = final_location - if final_attendees: - event_body["attendees"] = [ - {"email": e.strip()} for e in final_attendees if e.strip() - ] - - try: - created = await asyncio.get_event_loop().run_in_executor( - None, - lambda: ( - service.events() - .insert(calendarId="primary", body=event_body) - .execute() - ), + service = await asyncio.get_event_loop().run_in_executor( + None, lambda: build("calendar", "v3", credentials=creds) ) - except Exception as api_err: - from googleapiclient.errors import HttpError - if isinstance(api_err, HttpError) and api_err.resp.status == 403: - logger.warning( - f"Insufficient permissions for connector {actual_connector_id}: {api_err}" + event_body: dict[str, Any] = { + "summary": final_summary, + "start": {"dateTime": final_start_datetime, "timeZone": tz}, + "end": {"dateTime": final_end_datetime, "timeZone": tz}, + } + if final_description: + event_body["description"] = final_description + if final_location: + event_body["location"] = final_location + if final_attendees: + event_body["attendees"] = [ + {"email": e.strip()} for e in final_attendees if e.strip() + ] + + try: + created = await asyncio.get_event_loop().run_in_executor( + None, + lambda: ( + service.events() + .insert(calendarId="primary", body=event_body) + .execute() + ), ) - try: - from sqlalchemy.orm.attributes import flag_modified + except Exception as api_err: + from googleapiclient.errors import HttpError - _res = await db_session.execute( - select(SearchSourceConnector).where( - SearchSourceConnector.id == actual_connector_id - ) - ) - _conn = _res.scalar_one_or_none() - if _conn and not _conn.config.get("auth_expired"): - _conn.config = {**_conn.config, "auth_expired": True} - flag_modified(_conn, "config") - await db_session.commit() - except Exception: + if isinstance(api_err, HttpError) and api_err.resp.status == 403: logger.warning( - "Failed to persist auth_expired for connector %s", - actual_connector_id, - exc_info=True, + f"Insufficient permissions for connector {actual_connector_id}: {api_err}" ) - return { - "status": "insufficient_permissions", - "connector_id": actual_connector_id, - "message": "This Google Calendar account needs additional permissions. Please re-authenticate in connector settings.", - } - raise + try: + from sqlalchemy.orm.attributes import flag_modified - logger.info( - f"Calendar event created: id={created.get('id')}, summary={created.get('summary')}" - ) + _res = await db_session.execute( + select(SearchSourceConnector).where( + SearchSourceConnector.id == actual_connector_id + ) + ) + _conn = _res.scalar_one_or_none() + if _conn and not _conn.config.get("auth_expired"): + _conn.config = {**_conn.config, "auth_expired": True} + flag_modified(_conn, "config") + await db_session.commit() + except Exception: + logger.warning( + "Failed to persist auth_expired for connector %s", + actual_connector_id, + exc_info=True, + ) + return { + "status": "insufficient_permissions", + "connector_id": actual_connector_id, + "message": "This Google Calendar account needs additional permissions. Please re-authenticate in connector settings.", + } + raise + + logger.info( + f"Calendar event created via Google API: id={created.get('id')}, summary={created.get('summary')}" + ) kb_message_suffix = "" try: diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/delete_event.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/delete_event.py index 4d9d69b4b..3d160e669 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/delete_event.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/delete_event.py @@ -163,16 +163,22 @@ def create_delete_calendar_event_tool( connector.connector_type == SearchSourceConnectorType.COMPOSIO_GOOGLE_CALENDAR_CONNECTOR ): - from app.utils.google_credentials import build_composio_credentials - cca_id = connector.config.get("composio_connected_account_id") - if cca_id: - creds = build_composio_credentials(cca_id) - else: + if not cca_id: return { "status": "error", "message": "Composio connected account ID not found for this connector.", } + + from app.services.composio_service import ComposioService + + error = await ComposioService().delete_calendar_event( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + event_id=final_event_id, + ) + if error: + return {"status": "error", "message": error} else: config_data = dict(connector.config) @@ -202,51 +208,51 @@ def create_delete_calendar_event_tool( expiry=datetime.fromisoformat(exp) if exp else None, ) - service = await asyncio.get_event_loop().run_in_executor( - None, lambda: build("calendar", "v3", credentials=creds) - ) - - try: - await asyncio.get_event_loop().run_in_executor( - None, - lambda: ( - service.events() - .delete(calendarId="primary", eventId=final_event_id) - .execute() - ), + service = await asyncio.get_event_loop().run_in_executor( + None, lambda: build("calendar", "v3", credentials=creds) ) - except Exception as api_err: - from googleapiclient.errors import HttpError - if isinstance(api_err, HttpError) and api_err.resp.status == 403: - logger.warning( - f"Insufficient permissions for connector {actual_connector_id}: {api_err}" + try: + await asyncio.get_event_loop().run_in_executor( + None, + lambda: ( + service.events() + .delete(calendarId="primary", eventId=final_event_id) + .execute() + ), ) - try: - from sqlalchemy.orm.attributes import flag_modified + except Exception as api_err: + from googleapiclient.errors import HttpError - _res = await db_session.execute( - select(SearchSourceConnector).where( - SearchSourceConnector.id == actual_connector_id - ) - ) - _conn = _res.scalar_one_or_none() - if _conn and not _conn.config.get("auth_expired"): - _conn.config = {**_conn.config, "auth_expired": True} - flag_modified(_conn, "config") - await db_session.commit() - except Exception: + if isinstance(api_err, HttpError) and api_err.resp.status == 403: logger.warning( - "Failed to persist auth_expired for connector %s", - actual_connector_id, - exc_info=True, + f"Insufficient permissions for connector {actual_connector_id}: {api_err}" ) - return { - "status": "insufficient_permissions", - "connector_id": actual_connector_id, - "message": "This Google Calendar account needs additional permissions. Please re-authenticate in connector settings.", - } - raise + try: + from sqlalchemy.orm.attributes import flag_modified + + _res = await db_session.execute( + select(SearchSourceConnector).where( + SearchSourceConnector.id == actual_connector_id + ) + ) + _conn = _res.scalar_one_or_none() + if _conn and not _conn.config.get("auth_expired"): + _conn.config = {**_conn.config, "auth_expired": True} + flag_modified(_conn, "config") + await db_session.commit() + except Exception: + logger.warning( + "Failed to persist auth_expired for connector %s", + actual_connector_id, + exc_info=True, + ) + return { + "status": "insufficient_permissions", + "connector_id": actual_connector_id, + "message": "This Google Calendar account needs additional permissions. Please re-authenticate in connector settings.", + } + raise logger.info(f"Calendar event deleted: event_id={final_event_id}") diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/search_events.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/search_events.py index dc6adb822..6772d5a1e 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/search_events.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/search_events.py @@ -16,6 +16,14 @@ _CALENDAR_TYPES = [ ] +def _to_calendar_boundary(value: str, *, is_end: bool) -> str: + """Promote a bare YYYY-MM-DD to RFC3339 with a day-edge time, leave full datetimes alone.""" + if "T" in value: + return value + time = "23:59:59" if is_end else "00:00:00" + return f"{value}T{time}Z" + + def create_search_calendar_events_tool( db_session: AsyncSession | None = None, search_space_id: int | None = None, @@ -61,22 +69,47 @@ def create_search_calendar_events_tool( "message": "No Google Calendar connector found. Please connect Google Calendar in your workspace settings.", } - creds = _build_credentials(connector) + if ( + connector.connector_type + == SearchSourceConnectorType.COMPOSIO_GOOGLE_CALENDAR_CONNECTOR + ): + cca_id = connector.config.get("composio_connected_account_id") + if not cca_id: + return { + "status": "error", + "message": "Composio connected account ID not found for this connector.", + } - from app.connectors.google_calendar_connector import GoogleCalendarConnector + from app.services.composio_service import ComposioService - cal = GoogleCalendarConnector( - credentials=creds, - session=db_session, - user_id=user_id, - connector_id=connector.id, - ) + events_raw, error = await ComposioService().get_calendar_events( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + time_min=_to_calendar_boundary(start_date, is_end=False), + time_max=_to_calendar_boundary(end_date, is_end=True), + max_results=max_results, + ) + if not events_raw and not error: + error = "No events found in the specified date range." + else: + creds = _build_credentials(connector) - events_raw, error = await cal.get_all_primary_calendar_events( - start_date=start_date, - end_date=end_date, - max_results=max_results, - ) + from app.connectors.google_calendar_connector import ( + GoogleCalendarConnector, + ) + + cal = GoogleCalendarConnector( + credentials=creds, + session=db_session, + user_id=user_id, + connector_id=connector.id, + ) + + events_raw, error = await cal.get_all_primary_calendar_events( + start_date=start_date, + end_date=end_date, + max_results=max_results, + ) if error: if ( diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/update_event.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/update_event.py index 259f52bba..a74979484 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/update_event.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/update_event.py @@ -192,20 +192,62 @@ def create_update_calendar_event_tool( f"Updating calendar event: event_id='{final_event_id}', connector={actual_connector_id}" ) + has_changes = any( + v is not None + for v in ( + final_new_summary, + final_new_start_datetime, + final_new_end_datetime, + final_new_description, + final_new_location, + final_new_attendees, + ) + ) + if not has_changes: + return { + "status": "error", + "message": "No changes specified. Please provide at least one field to update.", + } + if ( connector.connector_type == SearchSourceConnectorType.COMPOSIO_GOOGLE_CALENDAR_CONNECTOR ): - from app.utils.google_credentials import build_composio_credentials - cca_id = connector.config.get("composio_connected_account_id") - if cca_id: - creds = build_composio_credentials(cca_id) - else: + if not cca_id: return { "status": "error", "message": "Composio connected account ID not found for this connector.", } + + from app.services.composio_service import ComposioService + + tz_for_composio: str | None = None + if final_new_start_datetime is not None and not _is_date_only( + final_new_start_datetime + ): + tz_for_composio = ( + context.get("timezone") if isinstance(context, dict) else None + ) + + _, html_link, error = await ComposioService().update_calendar_event( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + event_id=final_event_id, + summary=final_new_summary, + start_time=final_new_start_datetime, + end_time=final_new_end_datetime, + timezone=tz_for_composio, + description=final_new_description, + location=final_new_location, + attendees=final_new_attendees, + ) + if error: + return {"status": "error", "message": error} + updated = {"htmlLink": html_link} + logger.info( + f"Calendar event updated via Composio: event_id={final_event_id}" + ) else: config_data = dict(connector.config) @@ -235,81 +277,79 @@ def create_update_calendar_event_tool( expiry=datetime.fromisoformat(exp) if exp else None, ) - service = await asyncio.get_event_loop().run_in_executor( - None, lambda: build("calendar", "v3", credentials=creds) - ) - - update_body: dict[str, Any] = {} - if final_new_summary is not None: - update_body["summary"] = final_new_summary - if final_new_start_datetime is not None: - update_body["start"] = _build_time_body( - final_new_start_datetime, context + service = await asyncio.get_event_loop().run_in_executor( + None, lambda: build("calendar", "v3", credentials=creds) ) - if final_new_end_datetime is not None: - update_body["end"] = _build_time_body(final_new_end_datetime, context) - if final_new_description is not None: - update_body["description"] = final_new_description - if final_new_location is not None: - update_body["location"] = final_new_location - if final_new_attendees is not None: - update_body["attendees"] = [ - {"email": e.strip()} for e in final_new_attendees if e.strip() - ] - if not update_body: - return { - "status": "error", - "message": "No changes specified. Please provide at least one field to update.", - } - - try: - updated = await asyncio.get_event_loop().run_in_executor( - None, - lambda: ( - service.events() - .patch( - calendarId="primary", - eventId=final_event_id, - body=update_body, - ) - .execute() - ), - ) - except Exception as api_err: - from googleapiclient.errors import HttpError - - if isinstance(api_err, HttpError) and api_err.resp.status == 403: - logger.warning( - f"Insufficient permissions for connector {actual_connector_id}: {api_err}" + update_body: dict[str, Any] = {} + if final_new_summary is not None: + update_body["summary"] = final_new_summary + if final_new_start_datetime is not None: + update_body["start"] = _build_time_body( + final_new_start_datetime, context ) - try: - from sqlalchemy.orm.attributes import flag_modified + if final_new_end_datetime is not None: + update_body["end"] = _build_time_body( + final_new_end_datetime, context + ) + if final_new_description is not None: + update_body["description"] = final_new_description + if final_new_location is not None: + update_body["location"] = final_new_location + if final_new_attendees is not None: + update_body["attendees"] = [ + {"email": e.strip()} for e in final_new_attendees if e.strip() + ] - _res = await db_session.execute( - select(SearchSourceConnector).where( - SearchSourceConnector.id == actual_connector_id + try: + updated = await asyncio.get_event_loop().run_in_executor( + None, + lambda: ( + service.events() + .patch( + calendarId="primary", + eventId=final_event_id, + body=update_body, ) - ) - _conn = _res.scalar_one_or_none() - if _conn and not _conn.config.get("auth_expired"): - _conn.config = {**_conn.config, "auth_expired": True} - flag_modified(_conn, "config") - await db_session.commit() - except Exception: - logger.warning( - "Failed to persist auth_expired for connector %s", - actual_connector_id, - exc_info=True, - ) - return { - "status": "insufficient_permissions", - "connector_id": actual_connector_id, - "message": "This Google Calendar account needs additional permissions. Please re-authenticate in connector settings.", - } - raise + .execute() + ), + ) + except Exception as api_err: + from googleapiclient.errors import HttpError - logger.info(f"Calendar event updated: event_id={final_event_id}") + if isinstance(api_err, HttpError) and api_err.resp.status == 403: + logger.warning( + f"Insufficient permissions for connector {actual_connector_id}: {api_err}" + ) + try: + from sqlalchemy.orm.attributes import flag_modified + + _res = await db_session.execute( + select(SearchSourceConnector).where( + SearchSourceConnector.id == actual_connector_id + ) + ) + _conn = _res.scalar_one_or_none() + if _conn and not _conn.config.get("auth_expired"): + _conn.config = {**_conn.config, "auth_expired": True} + flag_modified(_conn, "config") + await db_session.commit() + except Exception: + logger.warning( + "Failed to persist auth_expired for connector %s", + actual_connector_id, + exc_info=True, + ) + return { + "status": "insufficient_permissions", + "connector_id": actual_connector_id, + "message": "This Google Calendar account needs additional permissions. Please re-authenticate in connector settings.", + } + raise + + logger.info( + f"Calendar event updated via Google API: event_id={final_event_id}" + ) kb_message_suffix = "" if document_id is not None: diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/create_draft.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/create_draft.py index 0bd044695..59e471097 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/create_draft.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/create_draft.py @@ -161,16 +161,39 @@ def create_create_gmail_draft_tool( connector.connector_type == SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR ): - from app.utils.google_credentials import build_composio_credentials - cca_id = connector.config.get("composio_connected_account_id") - if cca_id: - creds = build_composio_credentials(cca_id) - else: + if not cca_id: return { "status": "error", "message": "Composio connected account ID not found for this Gmail connector.", } + + from app.services.composio_service import ComposioService + + ( + draft_id, + draft_message_id, + draft_thread_id, + error, + ) = await ComposioService().create_gmail_draft( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + to=final_to, + subject=final_subject, + body=final_body, + cc=final_cc, + bcc=final_bcc, + ) + if error: + return {"status": "error", "message": error} + created = { + "id": draft_id, + "message": { + "id": draft_message_id, + "threadId": draft_thread_id, + }, + } + logger.info(f"Gmail draft created via Composio: id={draft_id}") else: from google.oauth2.credentials import Credentials @@ -208,63 +231,65 @@ def create_create_gmail_draft_tool( expiry=datetime.fromisoformat(exp) if exp else None, ) - from googleapiclient.discovery import build + from googleapiclient.discovery import build - gmail_service = build("gmail", "v1", credentials=creds) + gmail_service = build("gmail", "v1", credentials=creds) - message = MIMEText(final_body) - message["to"] = final_to - message["subject"] = final_subject - if final_cc: - message["cc"] = final_cc - if final_bcc: - message["bcc"] = final_bcc - raw = base64.urlsafe_b64encode(message.as_bytes()).decode() + message = MIMEText(final_body) + message["to"] = final_to + message["subject"] = final_subject + if final_cc: + message["cc"] = final_cc + if final_bcc: + message["bcc"] = final_bcc + raw = base64.urlsafe_b64encode(message.as_bytes()).decode() - try: - created = await asyncio.get_event_loop().run_in_executor( - None, - lambda: ( - gmail_service.users() - .drafts() - .create(userId="me", body={"message": {"raw": raw}}) - .execute() - ), - ) - except Exception as api_err: - from googleapiclient.errors import HttpError - - if isinstance(api_err, HttpError) and api_err.resp.status == 403: - logger.warning( - f"Insufficient permissions for connector {actual_connector_id}: {api_err}" + try: + created = await asyncio.get_event_loop().run_in_executor( + None, + lambda: ( + gmail_service.users() + .drafts() + .create(userId="me", body={"message": {"raw": raw}}) + .execute() + ), ) - try: - from sqlalchemy.orm.attributes import flag_modified + except Exception as api_err: + from googleapiclient.errors import HttpError - _res = await db_session.execute( - select(SearchSourceConnector).where( - SearchSourceConnector.id == actual_connector_id - ) - ) - _conn = _res.scalar_one_or_none() - if _conn and not _conn.config.get("auth_expired"): - _conn.config = {**_conn.config, "auth_expired": True} - flag_modified(_conn, "config") - await db_session.commit() - except Exception: + if isinstance(api_err, HttpError) and api_err.resp.status == 403: logger.warning( - "Failed to persist auth_expired for connector %s", - actual_connector_id, - exc_info=True, + f"Insufficient permissions for connector {actual_connector_id}: {api_err}" ) - return { - "status": "insufficient_permissions", - "connector_id": actual_connector_id, - "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.", - } - raise + try: + from sqlalchemy.orm.attributes import flag_modified - logger.info(f"Gmail draft created: id={created.get('id')}") + _res = await db_session.execute( + select(SearchSourceConnector).where( + SearchSourceConnector.id == actual_connector_id + ) + ) + _conn = _res.scalar_one_or_none() + if _conn and not _conn.config.get("auth_expired"): + _conn.config = {**_conn.config, "auth_expired": True} + flag_modified(_conn, "config") + await db_session.commit() + except Exception: + logger.warning( + "Failed to persist auth_expired for connector %s", + actual_connector_id, + exc_info=True, + ) + return { + "status": "insufficient_permissions", + "connector_id": actual_connector_id, + "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.", + } + raise + + logger.info( + f"Gmail draft created via Google API: id={created.get('id')}" + ) kb_message_suffix = "" try: diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/read_email.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/read_email.py index deec1627c..39526f25e 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/read_email.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/read_email.py @@ -50,7 +50,56 @@ def create_read_gmail_email_tool( "message": "No Gmail connector found. Please connect Gmail in your workspace settings.", } - from app.agents.new_chat.tools.gmail.search_emails import _build_credentials + if ( + connector.connector_type + == SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR + ): + cca_id = connector.config.get("composio_connected_account_id") + if not cca_id: + return { + "status": "error", + "message": "Composio connected account ID not found for this Gmail connector.", + } + + from app.agents.new_chat.tools.gmail.search_emails import ( + _format_gmail_summary, + ) + from app.services.composio_service import ComposioService + + detail, error = await ComposioService().get_gmail_message_detail( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + message_id=message_id, + ) + if error: + return {"status": "error", "message": error} + if not detail: + return { + "status": "not_found", + "message": f"Email with ID '{message_id}' not found.", + } + + summary = _format_gmail_summary(detail) + content = ( + f"# {summary['subject']}\n\n" + f"**From:** {summary['from']}\n" + f"**To:** {summary['to']}\n" + f"**Date:** {summary['date']}\n\n" + f"## Message Content\n\n" + f"{detail.get('messageText') or detail.get('snippet') or ''}\n\n" + f"## Message Details\n\n" + f"- **Message ID:** {summary['message_id']}\n" + f"- **Thread ID:** {summary['thread_id']}\n" + ) + return { + "status": "success", + "message_id": summary["message_id"] or message_id, + "content": content, + } + + from app.agents.new_chat.tools.gmail.search_emails import ( + _build_credentials, + ) creds = _build_credentials(connector) diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/search_emails.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/search_emails.py index 2e363609e..a9d7cdedf 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/search_emails.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/search_emails.py @@ -1,5 +1,4 @@ import logging -from datetime import datetime from typing import Any from langchain_core.tools import tool @@ -15,57 +14,6 @@ _GMAIL_TYPES = [ SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR, ] -_token_encryption_cache: object | None = None - - -def _get_token_encryption(): - global _token_encryption_cache - if _token_encryption_cache is None: - from app.config import config - from app.utils.oauth_security import TokenEncryption - - if not config.SECRET_KEY: - raise RuntimeError("SECRET_KEY not configured for token decryption.") - _token_encryption_cache = TokenEncryption(config.SECRET_KEY) - return _token_encryption_cache - - -def _build_credentials(connector: SearchSourceConnector): - """Build Google OAuth Credentials from a connector's stored config. - - Handles both native OAuth connectors (with encrypted tokens) and - Composio-backed connectors. Shared by Gmail and Calendar tools. - """ - from app.utils.google_credentials import COMPOSIO_GOOGLE_CONNECTOR_TYPES - - if connector.connector_type in COMPOSIO_GOOGLE_CONNECTOR_TYPES: - from app.utils.google_credentials import build_composio_credentials - - cca_id = connector.config.get("composio_connected_account_id") - if not cca_id: - raise ValueError("Composio connected account ID not found.") - return build_composio_credentials(cca_id) - - from google.oauth2.credentials import Credentials - - cfg = dict(connector.config) - if cfg.get("_token_encrypted"): - enc = _get_token_encryption() - for key in ("token", "refresh_token", "client_secret"): - if cfg.get(key): - cfg[key] = enc.decrypt_token(cfg[key]) - - exp = (cfg.get("expiry") or "").replace("Z", "") - return Credentials( - token=cfg.get("token"), - refresh_token=cfg.get("refresh_token"), - token_uri=cfg.get("token_uri"), - client_id=cfg.get("client_id"), - client_secret=cfg.get("client_secret"), - scopes=cfg.get("scopes", []), - expiry=datetime.fromisoformat(exp) if exp else None, - ) - def create_search_gmail_tool( db_session: AsyncSession | None = None, @@ -110,6 +58,50 @@ def create_search_gmail_tool( "message": "No Gmail connector found. Please connect Gmail in your workspace settings.", } + if ( + connector.connector_type + == SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR + ): + cca_id = connector.config.get("composio_connected_account_id") + if not cca_id: + return { + "status": "error", + "message": "Composio connected account ID not found for this Gmail connector.", + } + + from app.agents.new_chat.tools.gmail.search_emails import ( + _format_gmail_summary, + ) + from app.services.composio_service import ComposioService + + ( + messages, + _next, + _estimate, + error, + ) = await ComposioService().get_gmail_messages( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + query=query, + max_results=max_results, + ) + if error: + return {"status": "error", "message": error} + + emails = [_format_gmail_summary(m) for m in messages] + if not emails: + return { + "status": "success", + "emails": [], + "total": 0, + "message": "No emails found.", + } + return {"status": "success", "emails": emails, "total": len(emails)} + + from app.agents.new_chat.tools.gmail.search_emails import ( + _build_credentials, + ) + creds = _build_credentials(connector) from app.connectors.google_gmail_connector import GoogleGmailConnector diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/send_email.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/send_email.py index c3f0999f4..d5de24b62 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/send_email.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/send_email.py @@ -162,16 +162,31 @@ def create_send_gmail_email_tool( connector.connector_type == SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR ): - from app.utils.google_credentials import build_composio_credentials - cca_id = connector.config.get("composio_connected_account_id") - if cca_id: - creds = build_composio_credentials(cca_id) - else: + if not cca_id: return { "status": "error", "message": "Composio connected account ID not found for this Gmail connector.", } + + from app.services.composio_service import ComposioService + + ( + sent_message_id, + sent_thread_id, + error, + ) = await ComposioService().send_gmail_email( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + to=final_to, + subject=final_subject, + body=final_body, + cc=final_cc, + bcc=final_bcc, + ) + if error: + return {"status": "error", "message": error} + sent = {"id": sent_message_id, "threadId": sent_thread_id} else: from google.oauth2.credentials import Credentials @@ -209,61 +224,61 @@ def create_send_gmail_email_tool( expiry=datetime.fromisoformat(exp) if exp else None, ) - from googleapiclient.discovery import build + from googleapiclient.discovery import build - gmail_service = build("gmail", "v1", credentials=creds) + gmail_service = build("gmail", "v1", credentials=creds) - message = MIMEText(final_body) - message["to"] = final_to - message["subject"] = final_subject - if final_cc: - message["cc"] = final_cc - if final_bcc: - message["bcc"] = final_bcc - raw = base64.urlsafe_b64encode(message.as_bytes()).decode() + message = MIMEText(final_body) + message["to"] = final_to + message["subject"] = final_subject + if final_cc: + message["cc"] = final_cc + if final_bcc: + message["bcc"] = final_bcc + raw = base64.urlsafe_b64encode(message.as_bytes()).decode() - try: - sent = await asyncio.get_event_loop().run_in_executor( - None, - lambda: ( - gmail_service.users() - .messages() - .send(userId="me", body={"raw": raw}) - .execute() - ), - ) - except Exception as api_err: - from googleapiclient.errors import HttpError - - if isinstance(api_err, HttpError) and api_err.resp.status == 403: - logger.warning( - f"Insufficient permissions for connector {actual_connector_id}: {api_err}" + try: + sent = await asyncio.get_event_loop().run_in_executor( + None, + lambda: ( + gmail_service.users() + .messages() + .send(userId="me", body={"raw": raw}) + .execute() + ), ) - try: - from sqlalchemy.orm.attributes import flag_modified + except Exception as api_err: + from googleapiclient.errors import HttpError - _res = await db_session.execute( - select(SearchSourceConnector).where( - SearchSourceConnector.id == actual_connector_id - ) - ) - _conn = _res.scalar_one_or_none() - if _conn and not _conn.config.get("auth_expired"): - _conn.config = {**_conn.config, "auth_expired": True} - flag_modified(_conn, "config") - await db_session.commit() - except Exception: + if isinstance(api_err, HttpError) and api_err.resp.status == 403: logger.warning( - "Failed to persist auth_expired for connector %s", - actual_connector_id, - exc_info=True, + f"Insufficient permissions for connector {actual_connector_id}: {api_err}" ) - return { - "status": "insufficient_permissions", - "connector_id": actual_connector_id, - "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.", - } - raise + try: + from sqlalchemy.orm.attributes import flag_modified + + _res = await db_session.execute( + select(SearchSourceConnector).where( + SearchSourceConnector.id == actual_connector_id + ) + ) + _conn = _res.scalar_one_or_none() + if _conn and not _conn.config.get("auth_expired"): + _conn.config = {**_conn.config, "auth_expired": True} + flag_modified(_conn, "config") + await db_session.commit() + except Exception: + logger.warning( + "Failed to persist auth_expired for connector %s", + actual_connector_id, + exc_info=True, + ) + return { + "status": "insufficient_permissions", + "connector_id": actual_connector_id, + "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.", + } + raise logger.info( f"Gmail email sent: id={sent.get('id')}, threadId={sent.get('threadId')}" diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/trash_email.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/trash_email.py index 1f1f6227a..b78f88934 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/trash_email.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/trash_email.py @@ -162,16 +162,22 @@ def create_trash_gmail_email_tool( connector.connector_type == SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR ): - from app.utils.google_credentials import build_composio_credentials - cca_id = connector.config.get("composio_connected_account_id") - if cca_id: - creds = build_composio_credentials(cca_id) - else: + if not cca_id: return { "status": "error", "message": "Composio connected account ID not found for this Gmail connector.", } + + from app.services.composio_service import ComposioService + + error = await ComposioService().trash_gmail_message( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + message_id=final_message_id, + ) + if error: + return {"status": "error", "message": error} else: from google.oauth2.credentials import Credentials @@ -209,49 +215,49 @@ def create_trash_gmail_email_tool( expiry=datetime.fromisoformat(exp) if exp else None, ) - from googleapiclient.discovery import build + from googleapiclient.discovery import build - gmail_service = build("gmail", "v1", credentials=creds) + gmail_service = build("gmail", "v1", credentials=creds) - try: - await asyncio.get_event_loop().run_in_executor( - None, - lambda: ( - gmail_service.users() - .messages() - .trash(userId="me", id=final_message_id) - .execute() - ), - ) - except Exception as api_err: - from googleapiclient.errors import HttpError - - if isinstance(api_err, HttpError) and api_err.resp.status == 403: - logger.warning( - f"Insufficient permissions for connector {connector.id}: {api_err}" + try: + await asyncio.get_event_loop().run_in_executor( + None, + lambda: ( + gmail_service.users() + .messages() + .trash(userId="me", id=final_message_id) + .execute() + ), ) - try: - from sqlalchemy.orm.attributes import flag_modified + except Exception as api_err: + from googleapiclient.errors import HttpError - if not connector.config.get("auth_expired"): - connector.config = { - **connector.config, - "auth_expired": True, - } - flag_modified(connector, "config") - await db_session.commit() - except Exception: + if isinstance(api_err, HttpError) and api_err.resp.status == 403: logger.warning( - "Failed to persist auth_expired for connector %s", - connector.id, - exc_info=True, + f"Insufficient permissions for connector {connector.id}: {api_err}" ) - return { - "status": "insufficient_permissions", - "connector_id": connector.id, - "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.", - } - raise + try: + from sqlalchemy.orm.attributes import flag_modified + + if not connector.config.get("auth_expired"): + connector.config = { + **connector.config, + "auth_expired": True, + } + flag_modified(connector, "config") + await db_session.commit() + except Exception: + logger.warning( + "Failed to persist auth_expired for connector %s", + connector.id, + exc_info=True, + ) + return { + "status": "insufficient_permissions", + "connector_id": connector.id, + "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.", + } + raise logger.info(f"Gmail email trashed: message_id={final_message_id}") diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/update_draft.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/update_draft.py index 91178cd21..b6688ac53 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/update_draft.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/update_draft.py @@ -192,16 +192,51 @@ def create_update_gmail_draft_tool( connector.connector_type == SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR ): - from app.utils.google_credentials import build_composio_credentials - cca_id = connector.config.get("composio_connected_account_id") - if cca_id: - creds = build_composio_credentials(cca_id) - else: + if not cca_id: return { "status": "error", "message": "Composio connected account ID not found for this Gmail connector.", } + + if not final_draft_id: + return { + "status": "error", + "message": ( + "Could not find this draft in Gmail. " + "It may have already been sent or deleted." + ), + } + + from app.services.composio_service import ComposioService + + ( + new_draft_id, + new_message_id, + error, + ) = await ComposioService().update_gmail_draft( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + draft_id=final_draft_id, + to=final_to or None, + subject=final_subject, + body=final_body, + cc=final_cc, + bcc=final_bcc, + ) + if error: + if "not found" in error.lower() or "no longer" in error.lower(): + return { + "status": "error", + "message": "Draft no longer exists in Gmail. It may have been sent or deleted.", + } + return {"status": "error", "message": error} + + updated = { + "id": new_draft_id or final_draft_id, + "message": {"id": new_message_id} if new_message_id else {}, + } + logger.info(f"Gmail draft updated via Composio: id={updated.get('id')}") else: from google.oauth2.credentials import Credentials @@ -239,88 +274,90 @@ def create_update_gmail_draft_tool( expiry=datetime.fromisoformat(exp) if exp else None, ) - from googleapiclient.discovery import build + from googleapiclient.discovery import build - gmail_service = build("gmail", "v1", credentials=creds) + gmail_service = build("gmail", "v1", credentials=creds) - # Resolve draft_id if not already available - if not final_draft_id: - logger.info( - f"draft_id not in metadata, looking up via drafts.list for message_id={message_id}" - ) - final_draft_id = await _find_draft_id_by_message( - gmail_service, message_id - ) - - if not final_draft_id: - return { - "status": "error", - "message": ( - "Could not find this draft in Gmail. " - "It may have already been sent or deleted." - ), - } - - message = MIMEText(final_body) - if final_to: - message["to"] = final_to - message["subject"] = final_subject - if final_cc: - message["cc"] = final_cc - if final_bcc: - message["bcc"] = final_bcc - raw = base64.urlsafe_b64encode(message.as_bytes()).decode() - - try: - updated = await asyncio.get_event_loop().run_in_executor( - None, - lambda: ( - gmail_service.users() - .drafts() - .update( - userId="me", - id=final_draft_id, - body={"message": {"raw": raw}}, - ) - .execute() - ), - ) - except Exception as api_err: - from googleapiclient.errors import HttpError - - if isinstance(api_err, HttpError) and api_err.resp.status == 403: - logger.warning( - f"Insufficient permissions for connector {connector.id}: {api_err}" + # Resolve draft_id if not already available + if not final_draft_id: + logger.info( + f"draft_id not in metadata, looking up via drafts.list for message_id={message_id}" + ) + final_draft_id = await _find_draft_id_by_message( + gmail_service, message_id ) - try: - from sqlalchemy.orm.attributes import flag_modified - if not connector.config.get("auth_expired"): - connector.config = { - **connector.config, - "auth_expired": True, - } - flag_modified(connector, "config") - await db_session.commit() - except Exception: - logger.warning( - "Failed to persist auth_expired for connector %s", - connector.id, - exc_info=True, - ) - return { - "status": "insufficient_permissions", - "connector_id": connector.id, - "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.", - } - if isinstance(api_err, HttpError) and api_err.resp.status == 404: + if not final_draft_id: return { "status": "error", - "message": "Draft no longer exists in Gmail. It may have been sent or deleted.", + "message": ( + "Could not find this draft in Gmail. " + "It may have already been sent or deleted." + ), } - raise - logger.info(f"Gmail draft updated: id={updated.get('id')}") + message = MIMEText(final_body) + if final_to: + message["to"] = final_to + message["subject"] = final_subject + if final_cc: + message["cc"] = final_cc + if final_bcc: + message["bcc"] = final_bcc + raw = base64.urlsafe_b64encode(message.as_bytes()).decode() + + try: + updated = await asyncio.get_event_loop().run_in_executor( + None, + lambda: ( + gmail_service.users() + .drafts() + .update( + userId="me", + id=final_draft_id, + body={"message": {"raw": raw}}, + ) + .execute() + ), + ) + except Exception as api_err: + from googleapiclient.errors import HttpError + + if isinstance(api_err, HttpError) and api_err.resp.status == 403: + logger.warning( + f"Insufficient permissions for connector {connector.id}: {api_err}" + ) + try: + from sqlalchemy.orm.attributes import flag_modified + + if not connector.config.get("auth_expired"): + connector.config = { + **connector.config, + "auth_expired": True, + } + flag_modified(connector, "config") + await db_session.commit() + except Exception: + logger.warning( + "Failed to persist auth_expired for connector %s", + connector.id, + exc_info=True, + ) + return { + "status": "insufficient_permissions", + "connector_id": connector.id, + "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.", + } + if isinstance(api_err, HttpError) and api_err.resp.status == 404: + return { + "status": "error", + "message": "Draft no longer exists in Gmail. It may have been sent or deleted.", + } + raise + + logger.info( + f"Gmail draft updated via Google API: id={updated.get('id')}" + ) kb_message_suffix = "" if document_id: diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/google_drive/tools/create_file.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/google_drive/tools/create_file.py index f36db8f3f..9e9a30429 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/google_drive/tools/create_file.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/google_drive/tools/create_file.py @@ -179,59 +179,96 @@ def create_create_google_drive_file_tool( f"Creating Google Drive file: name='{final_name}', type='{final_file_type}', connector={actual_connector_id}" ) - pre_built_creds = None + async def _flag_auth_expired() -> None: + try: + from sqlalchemy.orm.attributes import flag_modified + + _res = await db_session.execute( + select(SearchSourceConnector).where( + SearchSourceConnector.id == actual_connector_id + ) + ) + _conn = _res.scalar_one_or_none() + if _conn and not _conn.config.get("auth_expired"): + _conn.config = {**_conn.config, "auth_expired": True} + flag_modified(_conn, "config") + await db_session.commit() + except Exception: + logger.warning( + "Failed to persist auth_expired for connector %s", + actual_connector_id, + exc_info=True, + ) + if ( connector.connector_type == SearchSourceConnectorType.COMPOSIO_GOOGLE_DRIVE_CONNECTOR ): - from app.utils.google_credentials import build_composio_credentials - cca_id = connector.config.get("composio_connected_account_id") - if cca_id: - pre_built_creds = build_composio_credentials(cca_id) + if not cca_id: + return { + "status": "error", + "message": "Composio connected account ID not found for this Google Drive connector.", + } - client = GoogleDriveClient( - session=db_session, - connector_id=actual_connector_id, - credentials=pre_built_creds, - ) - try: - created = await client.create_file( + from app.services.composio_service import ComposioService + + created, error = await ComposioService().create_drive_file_from_text( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", name=final_name, mime_type=mime_type, - parent_folder_id=final_parent_folder_id, content=final_content, + parent_id=final_parent_folder_id, ) - except HttpError as http_err: - if http_err.resp.status == 403: - logger.warning( - f"Insufficient permissions for connector {actual_connector_id}: {http_err}" - ) - try: - from sqlalchemy.orm.attributes import flag_modified - _res = await db_session.execute( - select(SearchSourceConnector).where( - SearchSourceConnector.id == actual_connector_id - ) - ) - _conn = _res.scalar_one_or_none() - if _conn and not _conn.config.get("auth_expired"): - _conn.config = {**_conn.config, "auth_expired": True} - flag_modified(_conn, "config") - await db_session.commit() - except Exception: + if error or not created: + err_lower = (error or "").lower() + if ( + "insufficient" in err_lower + or "permission" in err_lower + or "403" in err_lower + ): logger.warning( - "Failed to persist auth_expired for connector %s", - actual_connector_id, - exc_info=True, + f"Insufficient permissions for Composio Drive connector {actual_connector_id}: {error}" ) + await _flag_auth_expired() + return { + "status": "insufficient_permissions", + "connector_id": actual_connector_id, + "message": "This Google Drive account needs additional permissions. Please re-authenticate in connector settings.", + } + logger.error( + f"Composio Drive create_file failed for connector {actual_connector_id}: {error}" + ) return { - "status": "insufficient_permissions", - "connector_id": actual_connector_id, - "message": "This Google Drive account needs additional permissions. Please re-authenticate in connector settings.", + "status": "error", + "message": "Something went wrong while creating the file. Please try again.", } - raise + else: + client = GoogleDriveClient( + session=db_session, + connector_id=actual_connector_id, + ) + try: + created = await client.create_file( + name=final_name, + mime_type=mime_type, + parent_folder_id=final_parent_folder_id, + content=final_content, + ) + except HttpError as http_err: + if http_err.resp.status == 403: + logger.warning( + f"Insufficient permissions for connector {actual_connector_id}: {http_err}" + ) + await _flag_auth_expired() + return { + "status": "insufficient_permissions", + "connector_id": actual_connector_id, + "message": "This Google Drive account needs additional permissions. Please re-authenticate in connector settings.", + } + raise logger.info( f"Google Drive file created: id={created.get('id')}, name={created.get('name')}" diff --git a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/google_drive/tools/trash_file.py b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/google_drive/tools/trash_file.py index 832afff0d..f7531cf3d 100644 --- a/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/google_drive/tools/trash_file.py +++ b/surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/google_drive/tools/trash_file.py @@ -158,51 +158,84 @@ def create_delete_google_drive_file_tool( f"Deleting Google Drive file: file_id='{final_file_id}', connector={final_connector_id}" ) - pre_built_creds = None + async def _flag_auth_expired() -> None: + try: + from sqlalchemy.orm.attributes import flag_modified + + if not connector.config.get("auth_expired"): + connector.config = { + **connector.config, + "auth_expired": True, + } + flag_modified(connector, "config") + await db_session.commit() + except Exception: + logger.warning( + "Failed to persist auth_expired for connector %s", + connector.id, + exc_info=True, + ) + if ( connector.connector_type == SearchSourceConnectorType.COMPOSIO_GOOGLE_DRIVE_CONNECTOR ): - from app.utils.google_credentials import build_composio_credentials - cca_id = connector.config.get("composio_connected_account_id") - if cca_id: - pre_built_creds = build_composio_credentials(cca_id) - - client = GoogleDriveClient( - session=db_session, - connector_id=connector.id, - credentials=pre_built_creds, - ) - try: - await client.trash_file(file_id=final_file_id) - except HttpError as http_err: - if http_err.resp.status == 403: - logger.warning( - f"Insufficient permissions for connector {connector.id}: {http_err}" - ) - try: - from sqlalchemy.orm.attributes import flag_modified - - if not connector.config.get("auth_expired"): - connector.config = { - **connector.config, - "auth_expired": True, - } - flag_modified(connector, "config") - await db_session.commit() - except Exception: - logger.warning( - "Failed to persist auth_expired for connector %s", - connector.id, - exc_info=True, - ) + if not cca_id: return { - "status": "insufficient_permissions", - "connector_id": connector.id, - "message": "This Google Drive account needs additional permissions. Please re-authenticate in connector settings.", + "status": "error", + "message": "Composio connected account ID not found for this Google Drive connector.", } - raise + + from app.services.composio_service import ComposioService + + error = await ComposioService().trash_drive_file( + connected_account_id=cca_id, + entity_id=f"surfsense_{user_id}", + file_id=final_file_id, + ) + if error: + err_lower = error.lower() + if ( + "insufficient" in err_lower + or "permission" in err_lower + or "403" in err_lower + ): + logger.warning( + f"Insufficient permissions for Composio Drive connector {connector.id}: {error}" + ) + await _flag_auth_expired() + return { + "status": "insufficient_permissions", + "connector_id": connector.id, + "message": "This Google Drive account needs additional permissions. Please re-authenticate in connector settings.", + } + logger.error( + f"Composio Drive trash_file failed for connector {connector.id}: {error}" + ) + return { + "status": "error", + "message": "Something went wrong while trashing the file. Please try again.", + } + else: + client = GoogleDriveClient( + session=db_session, + connector_id=connector.id, + ) + try: + await client.trash_file(file_id=final_file_id) + except HttpError as http_err: + if http_err.resp.status == 403: + logger.warning( + f"Insufficient permissions for connector {connector.id}: {http_err}" + ) + await _flag_auth_expired() + return { + "status": "insufficient_permissions", + "connector_id": connector.id, + "message": "This Google Drive account needs additional permissions. Please re-authenticate in connector settings.", + } + raise logger.info( f"Google Drive file deleted (moved to trash): file_id={final_file_id}" diff --git a/surfsense_backend/app/services/composio_service.py b/surfsense_backend/app/services/composio_service.py index edfab1d15..d73a0d4ce 100644 --- a/surfsense_backend/app/services/composio_service.py +++ b/surfsense_backend/app/services/composio_service.py @@ -1027,6 +1027,505 @@ class ComposioService: logger.error(f"Failed to list Calendar events: {e!s}") return [], str(e) + @staticmethod + def _unwrap_response_data(data: Any) -> Any: + """Composio responses often nest the meaningful payload under + ``data.data.response_data``. Walk that envelope safely and return + whichever inner dict actually has the result keys.""" + if not isinstance(data, dict): + return data + inner = data.get("data", data) + if isinstance(inner, dict): + return inner.get("response_data", inner) + return inner + + @staticmethod + def _split_email_csv(value: str | None) -> list[str] | None: + """Tools accept comma-separated cc/bcc strings; Composio expects an array.""" + if not value: + return None + addrs = [e.strip() for e in value.split(",") if e.strip()] + return addrs or None + + # ===== Gmail write methods ===== + + async def send_gmail_email( + self, + connected_account_id: str, + entity_id: str, + to: str, + subject: str, + body: str, + cc: str | None = None, + bcc: str | None = None, + is_html: bool = False, + ) -> tuple[str | None, str | None, str | None]: + """Send a Gmail message via the Composio ``GMAIL_SEND_EMAIL`` toolkit. + + Returns: + Tuple of (message_id, thread_id, error). On success ``error`` is + None and at least one of the IDs is populated when Composio + returns them; on failure both IDs are None. + """ + try: + params: dict[str, Any] = { + "recipient_email": to, + "subject": subject, + "body": body, + "is_html": is_html, + } + if cc: + cc_list = self._split_email_csv(cc) + if cc_list: + params["cc"] = cc_list + if bcc: + bcc_list = self._split_email_csv(bcc) + if bcc_list: + params["bcc"] = bcc_list + + result = await self.execute_tool( + connected_account_id=connected_account_id, + tool_name="GMAIL_SEND_EMAIL", + params=params, + entity_id=entity_id, + ) + if not result.get("success"): + return None, None, result.get("error", "Unknown error") + + payload = self._unwrap_response_data(result.get("data", {})) + message_id = None + thread_id = None + if isinstance(payload, dict): + message_id = ( + payload.get("id") + or payload.get("message_id") + or payload.get("messageId") + ) + thread_id = payload.get("threadId") or payload.get("thread_id") + return message_id, thread_id, None + except Exception as e: + logger.error(f"Failed to send Gmail email: {e!s}") + return None, None, str(e) + + async def create_gmail_draft( + self, + connected_account_id: str, + entity_id: str, + to: str, + subject: str, + body: str, + cc: str | None = None, + bcc: str | None = None, + is_html: bool = False, + ) -> tuple[str | None, str | None, str | None, str | None]: + """Create a Gmail draft via the Composio ``GMAIL_CREATE_EMAIL_DRAFT`` toolkit. + + Returns: + Tuple of (draft_id, message_id, thread_id, error). On success + ``error`` is None and ``draft_id`` is populated. + """ + try: + params: dict[str, Any] = { + "recipient_email": to, + "subject": subject, + "body": body, + "is_html": is_html, + } + cc_list = self._split_email_csv(cc) + if cc_list: + params["cc"] = cc_list + bcc_list = self._split_email_csv(bcc) + if bcc_list: + params["bcc"] = bcc_list + + result = await self.execute_tool( + connected_account_id=connected_account_id, + tool_name="GMAIL_CREATE_EMAIL_DRAFT", + params=params, + entity_id=entity_id, + ) + if not result.get("success"): + return None, None, None, result.get("error", "Unknown error") + + payload = self._unwrap_response_data(result.get("data", {})) + draft_id = None + message_id = None + thread_id = None + if isinstance(payload, dict): + draft_id = payload.get("id") or payload.get("draft_id") + draft_message = payload.get("message") or {} + if isinstance(draft_message, dict): + message_id = draft_message.get("id") or draft_message.get( + "message_id" + ) + thread_id = draft_message.get("threadId") or draft_message.get( + "thread_id" + ) + if message_id is None: + message_id = payload.get("message_id") or payload.get("messageId") + if thread_id is None: + thread_id = payload.get("thread_id") or payload.get("threadId") + return draft_id, message_id, thread_id, None + except Exception as e: + logger.error(f"Failed to create Gmail draft: {e!s}") + return None, None, None, str(e) + + async def update_gmail_draft( + self, + connected_account_id: str, + entity_id: str, + draft_id: str, + to: str | None = None, + subject: str | None = None, + body: str | None = None, + cc: str | None = None, + bcc: str | None = None, + is_html: bool = False, + ) -> tuple[str | None, str | None, str | None]: + """Update an existing Gmail draft via ``GMAIL_UPDATE_DRAFT``. + + Returns: + Tuple of (draft_id, message_id, error). + """ + try: + params: dict[str, Any] = { + "draft_id": draft_id, + "is_html": is_html, + } + if to: + params["recipient_email"] = to + if subject is not None: + params["subject"] = subject + if body is not None: + params["body"] = body + cc_list = self._split_email_csv(cc) + if cc_list: + params["cc"] = cc_list + bcc_list = self._split_email_csv(bcc) + if bcc_list: + params["bcc"] = bcc_list + + result = await self.execute_tool( + connected_account_id=connected_account_id, + tool_name="GMAIL_UPDATE_DRAFT", + params=params, + entity_id=entity_id, + ) + if not result.get("success"): + return None, None, result.get("error", "Unknown error") + + payload = self._unwrap_response_data(result.get("data", {})) + new_draft_id = draft_id + message_id = None + if isinstance(payload, dict): + new_draft_id = payload.get("id") or payload.get("draft_id") or draft_id + draft_message = payload.get("message") or {} + if isinstance(draft_message, dict): + message_id = draft_message.get("id") or draft_message.get( + "message_id" + ) + if message_id is None: + message_id = payload.get("message_id") or payload.get("messageId") + return new_draft_id, message_id, None + except Exception as e: + logger.error(f"Failed to update Gmail draft: {e!s}") + return None, None, str(e) + + async def trash_gmail_message( + self, + connected_account_id: str, + entity_id: str, + message_id: str, + ) -> str | None: + """Move a Gmail message to trash via ``GMAIL_MOVE_TO_TRASH``. + + Returns the error message on failure, ``None`` on success. + """ + try: + result = await self.execute_tool( + connected_account_id=connected_account_id, + tool_name="GMAIL_MOVE_TO_TRASH", + params={"message_id": message_id}, + entity_id=entity_id, + ) + if not result.get("success"): + return result.get("error", "Unknown error") + return None + except Exception as e: + logger.error(f"Failed to trash Gmail message: {e!s}") + return str(e) + + # ===== Google Calendar write methods ===== + + async def create_calendar_event( + self, + connected_account_id: str, + entity_id: str, + summary: str, + start_datetime: str, + end_datetime: str, + timezone: str | None = None, + description: str | None = None, + location: str | None = None, + attendees: list[str] | None = None, + calendar_id: str = "primary", + ) -> tuple[str | None, str | None, str | None]: + """Create a Google Calendar event via ``GOOGLECALENDAR_CREATE_EVENT``. + + Composio strips trailing timezone info on ``start_datetime`` / + ``end_datetime`` and uses the ``timezone`` field as the IANA name, + so callers may pass ISO 8601 strings with or without offsets. + + Returns: + Tuple of (event_id, html_link, error). + """ + try: + params: dict[str, Any] = { + "summary": summary, + "start_datetime": start_datetime, + "end_datetime": end_datetime, + "calendar_id": calendar_id, + } + if timezone: + params["timezone"] = timezone + if description: + params["description"] = description + if location: + params["location"] = location + if attendees: + params["attendees"] = [a for a in attendees if a] + + result = await self.execute_tool( + connected_account_id=connected_account_id, + tool_name="GOOGLECALENDAR_CREATE_EVENT", + params=params, + entity_id=entity_id, + ) + if not result.get("success"): + return None, None, result.get("error", "Unknown error") + + payload = self._unwrap_response_data(result.get("data", {})) + event_id = None + html_link = None + if isinstance(payload, dict): + event_id = payload.get("id") or payload.get("event_id") + html_link = payload.get("htmlLink") or payload.get("html_link") + return event_id, html_link, None + except Exception as e: + logger.error(f"Failed to create Calendar event: {e!s}") + return None, None, str(e) + + async def update_calendar_event( + self, + connected_account_id: str, + entity_id: str, + event_id: str, + summary: str | None = None, + start_time: str | None = None, + end_time: str | None = None, + timezone: str | None = None, + description: str | None = None, + location: str | None = None, + attendees: list[str] | None = None, + calendar_id: str = "primary", + ) -> tuple[str | None, str | None, str | None]: + """Patch an existing Google Calendar event via ``GOOGLECALENDAR_PATCH_EVENT``. + + Uses PATCH (not PUT) semantics so omitted fields are preserved. + + Returns: + Tuple of (event_id, html_link, error). + """ + try: + params: dict[str, Any] = { + "event_id": event_id, + "calendar_id": calendar_id, + } + if summary is not None: + params["summary"] = summary + if start_time is not None: + params["start_time"] = start_time + if end_time is not None: + params["end_time"] = end_time + if timezone: + params["timezone"] = timezone + if description is not None: + params["description"] = description + if location is not None: + params["location"] = location + if attendees is not None: + params["attendees"] = [a for a in attendees if a] + + result = await self.execute_tool( + connected_account_id=connected_account_id, + tool_name="GOOGLECALENDAR_PATCH_EVENT", + params=params, + entity_id=entity_id, + ) + if not result.get("success"): + return None, None, result.get("error", "Unknown error") + + payload = self._unwrap_response_data(result.get("data", {})) + new_event_id = event_id + html_link = None + if isinstance(payload, dict): + new_event_id = payload.get("id") or payload.get("event_id") or event_id + html_link = payload.get("htmlLink") or payload.get("html_link") + return new_event_id, html_link, None + except Exception as e: + logger.error(f"Failed to patch Calendar event: {e!s}") + return None, None, str(e) + + async def delete_calendar_event( + self, + connected_account_id: str, + entity_id: str, + event_id: str, + calendar_id: str = "primary", + ) -> str | None: + """Delete a Google Calendar event via ``GOOGLECALENDAR_DELETE_EVENT``. + + Returns the error message on failure, ``None`` on success (idempotent + on already-deleted events). + """ + try: + result = await self.execute_tool( + connected_account_id=connected_account_id, + tool_name="GOOGLECALENDAR_DELETE_EVENT", + params={ + "event_id": event_id, + "calendar_id": calendar_id, + }, + entity_id=entity_id, + ) + if not result.get("success"): + return result.get("error", "Unknown error") + return None + except Exception as e: + logger.error(f"Failed to delete Calendar event: {e!s}") + return str(e) + + # ===== Google Drive write methods ===== + + @staticmethod + def _drive_web_view_link(file_id: str, mime_type: str | None) -> str: + """Synthesize a Google Drive ``webViewLink`` from id + mimeType. + + Composio's ``GOOGLEDRIVE_CREATE_FILE_FROM_TEXT`` returns flat + metadata (id, name, mimeType) but does not always include a + ``webViewLink``. We rebuild the canonical UI URL based on the + Workspace MIME type so callers can keep using a single field. + """ + if not file_id: + return "" + mt = (mime_type or "").lower() + if mt == "application/vnd.google-apps.document": + return f"https://docs.google.com/document/d/{file_id}/edit" + if mt == "application/vnd.google-apps.spreadsheet": + return f"https://docs.google.com/spreadsheets/d/{file_id}/edit" + if mt == "application/vnd.google-apps.presentation": + return f"https://docs.google.com/presentation/d/{file_id}/edit" + if mt == "application/vnd.google-apps.folder": + return f"https://drive.google.com/drive/folders/{file_id}" + return f"https://drive.google.com/file/d/{file_id}/view" + + async def create_drive_file_from_text( + self, + connected_account_id: str, + entity_id: str, + name: str, + mime_type: str, + content: str | None = None, + parent_id: str | None = None, + ) -> tuple[dict[str, Any] | None, str | None]: + """Create a Google Drive file from text via ``GOOGLEDRIVE_CREATE_FILE_FROM_TEXT``. + + Composio's tool requires ``text_content`` even for "empty" files; + an empty string is accepted. Native Workspace types (Docs, Sheets) + are produced by setting ``mime_type`` to the Google Apps MIME, and + Drive auto-converts the text payload (e.g. CSV → Sheet). + + Returns: + Tuple of (file_meta, error). ``file_meta`` keys: + ``id``, ``name``, ``mimeType``, ``webViewLink``. + """ + try: + params: dict[str, Any] = { + "file_name": name, + "mime_type": mime_type, + "text_content": content if content is not None else "", + } + if parent_id: + params["parent_id"] = parent_id + + result = await self.execute_tool( + connected_account_id=connected_account_id, + tool_name="GOOGLEDRIVE_CREATE_FILE_FROM_TEXT", + params=params, + entity_id=entity_id, + ) + if not result.get("success"): + return None, result.get("error", "Unknown error") + + payload = self._unwrap_response_data(result.get("data", {})) + file_id: str | None = None + file_name: str | None = name + mime: str | None = mime_type + web_view_link: str | None = None + + if isinstance(payload, dict): + file_id = ( + payload.get("id") or payload.get("file_id") or payload.get("fileId") + ) + file_name = payload.get("name") or payload.get("file_name") or name + mime = payload.get("mimeType") or payload.get("mime_type") or mime_type + web_view_link = payload.get("webViewLink") or payload.get( + "web_view_link" + ) + + if not file_id: + return None, "Composio response did not include a file id" + + if not web_view_link: + web_view_link = self._drive_web_view_link(file_id, mime) + + return ( + { + "id": file_id, + "name": file_name, + "mimeType": mime, + "webViewLink": web_view_link, + }, + None, + ) + except Exception as e: + logger.error(f"Failed to create Drive file: {e!s}") + return None, str(e) + + async def trash_drive_file( + self, + connected_account_id: str, + entity_id: str, + file_id: str, + ) -> str | None: + """Move a Google Drive file to trash via ``GOOGLEDRIVE_TRASH_FILE``. + + Returns the error message on failure, ``None`` on success. + """ + try: + result = await self.execute_tool( + connected_account_id=connected_account_id, + tool_name="GOOGLEDRIVE_TRASH_FILE", + params={"file_id": file_id}, + entity_id=entity_id, + ) + if not result.get("success"): + return result.get("error", "Unknown error") + return None + except Exception as e: + logger.error(f"Failed to trash Drive file: {e!s}") + return str(e) + # ===== User Info Methods ===== async def get_connected_account_email(