apunkt/SurfSense
1
0
Fork 0
mirror of https://github.com/MODSetter/SurfSense.git synced 2026-05-07 06:42:39 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: composio tool calls in composio connectors

Browse source
This commit is contained in:
DESKTOP-RTLN3BA\$punk 2026-05-05 18:57:10 -07:00
parent 4e174f17f2
commit 5e87a7a251
13 changed files with 1347 additions and 550 deletions
Download patch file Download diff file Expand all files Collapse all files

149
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/create_event.py
View file

@ -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:

94
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/delete_event.py
View file

@ -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}")

59
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/search_events.py
View file

@ -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 (

188
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/calendar/tools/update_event.py
View file

@ -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:

133
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/create_draft.py
View file

@ -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:

51
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/read_email.py
View file

@ -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)

96
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/search_emails.py
View file

@ -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

121
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/send_email.py
View file

@ -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')}"

90
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/trash_email.py
View file

@ -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}")

195
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/gmail/tools/update_draft.py
View file

@ -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:

113
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/google_drive/tools/create_file.py
View file

@ -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')}"

109
surfsense_backend/app/agents/multi_agent_chat/subagents/connectors/google_drive/tools/trash_file.py
View file

@ -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}"

499
surfsense_backend/app/services/composio_service.py
View file

@ -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(