apunkt/SurfSense
1
0
Fork 0
mirror of https://github.com/MODSetter/SurfSense.git synced 2026-05-21 18:55:16 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge commit 'e1e4bb4706' into dev_mod

Browse source
This commit is contained in:
DESKTOP-RTLN3BA\$punk 2026-04-13 20:35:04 -07:00
commit 5d3142332b
85 changed files with 2357 additions and 3132 deletions
Download patch file Download diff file Expand all files Collapse all files

2
surfsense_backend/app/agents/new_chat/chat_deepagent.py
View file

@ -472,7 +472,7 @@ async def create_surfsense_deep_agent(
SubAgentMiddleware(backend=StateBackend, subagents=[general_purpose_spec]),
create_summarization_middleware(llm, StateBackend),
PatchToolCallsMiddleware(),
DedupHITLToolCallsMiddleware(),
DedupHITLToolCallsMiddleware(agent_tools=tools),
AnthropicPromptCachingMiddleware(unsupported_model_behavior="ignore"),
]

56
surfsense_backend/app/agents/new_chat/middleware/dedup_tool_calls.py
View file

@ -20,19 +20,39 @@ from langgraph.runtime import Runtime
logger = logging.getLogger(__name__)
_HITL_TOOL_DEDUP_KEYS: dict[str, str] = {
"delete_calendar_event": "event_title_or_id",
"update_calendar_event": "event_title_or_id",
"trash_gmail_email": "email_subject_or_id",
_NATIVE_HITL_TOOL_DEDUP_KEYS: dict[str, str] = {
# Gmail
"send_gmail_email": "subject",
"create_gmail_draft": "subject",
"update_gmail_draft": "draft_subject_or_id",
"trash_gmail_email": "email_subject_or_id",
# Google Calendar
"create_calendar_event": "title",
"update_calendar_event": "event_title_or_id",
"delete_calendar_event": "event_title_or_id",
# Google Drive
"create_google_drive_file": "file_name",
"delete_google_drive_file": "file_name",
# OneDrive
"create_onedrive_file": "file_name",
"delete_onedrive_file": "file_name",
"delete_notion_page": "page_title",
# Dropbox
"create_dropbox_file": "file_name",
"delete_dropbox_file": "file_name",
# Notion
"create_notion_page": "title",
"update_notion_page": "page_title",
"delete_linear_issue": "issue_ref",
"delete_notion_page": "page_title",
# Linear
"create_linear_issue": "title",
"update_linear_issue": "issue_ref",
"delete_linear_issue": "issue_ref",
# Jira
"create_jira_issue": "summary",
"update_jira_issue": "issue_title_or_key",
"delete_jira_issue": "issue_title_or_key",
# Confluence
"create_confluence_page": "title",
"update_confluence_page": "page_title_or_id",
"delete_confluence_page": "page_title_or_id",
}
@ -43,22 +63,38 @@ class DedupHITLToolCallsMiddleware(AgentMiddleware): # type: ignore[type-arg]
Only the **first** occurrence of each (tool-name, primary-arg-value)
pair is kept; subsequent duplicates are silently dropped.
The dedup map is built from two sources:
1. A comprehensive list of native HITL tools (hardcoded above).
2. Any ``StructuredTool`` instances passed via *agent_tools* whose
``metadata`` contains ``{"hitl": True, "hitl_dedup_key": "..."}``.
This is how MCP tools automatically get dedup support.
"""
tools = ()
def __init__(self, *, agent_tools: list[Any] | None = None) -> None:
self._dedup_keys: dict[str, str] = dict(_NATIVE_HITL_TOOL_DEDUP_KEYS)
for t in agent_tools or []:
meta = getattr(t, "metadata", None) or {}
if meta.get("hitl") and meta.get("hitl_dedup_key"):
self._dedup_keys[t.name] = meta["hitl_dedup_key"]
def after_model(
self, state: AgentState, runtime: Runtime[Any]
) -> dict[str, Any] | None:
return self._dedup(state)
return self._dedup(state, self._dedup_keys)
async def aafter_model(
self, state: AgentState, runtime: Runtime[Any]
) -> dict[str, Any] | None:
return self._dedup(state)
return self._dedup(state, self._dedup_keys)
@staticmethod
def _dedup(state: AgentState) -> dict[str, Any] | None: # type: ignore[type-arg]
def _dedup(
state: AgentState, dedup_keys: dict[str, str] # type: ignore[type-arg]
) -> dict[str, Any] | None:
messages = state.get("messages")
if not messages:
return None
@ -73,7 +109,7 @@ class DedupHITLToolCallsMiddleware(AgentMiddleware): # type: ignore[type-arg]
for tc in tool_calls:
name = tc.get("name", "")
dedup_key_arg = _HITL_TOOL_DEDUP_KEYS.get(name)
dedup_key_arg = dedup_keys.get(name)
if dedup_key_arg is not None:
arg_val = str(tc.get("args", {}).get(dedup_key_arg, "")).lower()
key = (name, arg_val)

60
surfsense_backend/app/agents/new_chat/tools/confluence/create_page.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified
@ -65,54 +65,28 @@ def create_create_confluence_page_tool(
"connector_type": "confluence",
}
approval = interrupt(
{
"type": "confluence_page_creation",
"action": {
"tool": "create_confluence_page",
"params": {
"title": title,
"content": content,
"space_id": space_id,
"connector_id": connector_id,
},
},
"context": context,
}
result = request_approval(
action_type="confluence_page_creation",
tool_name="create_confluence_page",
params={
"title": title,
"content": content,
"space_id": space_id,
"connector_id": connector_id,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The page was not created.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_title = final_params.get("title", title)
final_content = final_params.get("content", content) or ""
final_space_id = final_params.get("space_id", space_id)
final_connector_id = final_params.get("connector_id", connector_id)
final_title = result.params.get("title", title)
final_content = result.params.get("content", content) or ""
final_space_id = result.params.get("space_id", space_id)
final_connector_id = result.params.get("connector_id", connector_id)
if not final_title or not final_title.strip():
return {"status": "error", "message": "Page title cannot be empty."}

56
surfsense_backend/app/agents/new_chat/tools/confluence/delete_page.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified
@ -74,54 +74,28 @@ def create_delete_confluence_page_tool(
document_id = page_data["document_id"]
connector_id_from_context = context.get("account", {}).get("id")
approval = interrupt(
{
"type": "confluence_page_deletion",
"action": {
"tool": "delete_confluence_page",
"params": {
"page_id": page_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
},
"context": context,
}
result = request_approval(
action_type="confluence_page_deletion",
tool_name="delete_confluence_page",
params={
"page_id": page_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The page was not deleted.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_page_id = final_params.get("page_id", page_id)
final_connector_id = final_params.get(
final_page_id = result.params.get("page_id", page_id)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_delete_from_kb = final_params.get("delete_from_kb", delete_from_kb)
final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
from sqlalchemy.future import select

68
surfsense_backend/app/agents/new_chat/tools/confluence/update_page.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified
@ -78,62 +78,36 @@ def create_update_confluence_page_tool(
document_id = page_data.get("document_id")
connector_id_from_context = context.get("account", {}).get("id")
approval = interrupt(
{
"type": "confluence_page_update",
"action": {
"tool": "update_confluence_page",
"params": {
"page_id": page_id,
"document_id": document_id,
"new_title": new_title,
"new_content": new_content,
"version": current_version,
"connector_id": connector_id_from_context,
},
},
"context": context,
}
result = request_approval(
action_type="confluence_page_update",
tool_name="update_confluence_page",
params={
"page_id": page_id,
"document_id": document_id,
"new_title": new_title,
"new_content": new_content,
"version": current_version,
"connector_id": connector_id_from_context,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The page was not updated.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_page_id = final_params.get("page_id", page_id)
final_title = final_params.get("new_title", new_title) or current_title
final_content = final_params.get("new_content", new_content)
final_page_id = result.params.get("page_id", page_id)
final_title = result.params.get("new_title", new_title) or current_title
final_content = result.params.get("new_content", new_content)
if final_content is None:
final_content = current_body
final_version = final_params.get("version", current_version)
final_connector_id = final_params.get(
final_version = result.params.get("version", current_version)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_document_id = final_params.get("document_id", document_id)
final_document_id = result.params.get("document_id", document_id)
from sqlalchemy.future import select

64
surfsense_backend/app/agents/new_chat/tools/dropbox/create_file.py
View file

@ -5,7 +5,7 @@ from pathlib import Path
from typing import Any, Literal
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
@ -159,56 +159,30 @@ def create_create_dropbox_file_tool(
"supported_types": _SUPPORTED_TYPES,
}
approval = interrupt(
{
"type": "dropbox_file_creation",
"action": {
"tool": "create_dropbox_file",
"params": {
"name": name,
"file_type": file_type,
"content": content,
"connector_id": None,
"parent_folder_path": None,
},
},
"context": context,
}
result = request_approval(
action_type="dropbox_file_creation",
tool_name="create_dropbox_file",
params={
"name": name,
"file_type": file_type,
"content": content,
"connector_id": None,
"parent_folder_path": None,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The file was not created.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_name = final_params.get("name", name)
final_file_type = final_params.get("file_type", file_type)
final_content = final_params.get("content", content)
final_connector_id = final_params.get("connector_id")
final_parent_folder_path = final_params.get("parent_folder_path")
final_name = result.params.get("name", name)
final_file_type = result.params.get("file_type", file_type)
final_content = result.params.get("content", content)
final_connector_id = result.params.get("connector_id")
final_parent_folder_path = result.params.get("parent_folder_path")
if not final_name or not final_name.strip():
return {"status": "error", "message": "File name cannot be empty."}

57
surfsense_backend/app/agents/new_chat/tools/dropbox/trash_file.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy import String, and_, cast, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
@ -174,53 +174,26 @@ def create_delete_dropbox_file_tool(
},
}
approval = interrupt(
{
"type": "dropbox_file_trash",
"action": {
"tool": "delete_dropbox_file",
"params": {
"file_path": file_path,
"connector_id": connector.id,
"delete_from_kb": delete_from_kb,
},
},
"context": context,
}
result = request_approval(
action_type="dropbox_file_trash",
tool_name="delete_dropbox_file",
params={
"file_path": file_path,
"connector_id": connector.id,
"delete_from_kb": delete_from_kb,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The file was not deleted. Do not ask again or suggest alternatives.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_file_path = final_params.get("file_path", file_path)
final_connector_id = final_params.get("connector_id", connector.id)
final_delete_from_kb = final_params.get("delete_from_kb", delete_from_kb)
final_file_path = result.params.get("file_path", file_path)
final_connector_id = result.params.get("connector_id", connector.id)
final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
if final_connector_id != connector.id:
result = await db_session.execute(

68
surfsense_backend/app/agents/new_chat/tools/gmail/create_draft.py
View file

@ -6,7 +6,7 @@ from email.mime.text import MIMEText
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.gmail import GmailToolMetadataService
@ -85,60 +85,32 @@ def create_create_gmail_draft_tool(
logger.info(
f"Requesting approval for creating Gmail draft: to='{to}', subject='{subject}'"
)
approval = interrupt(
{
"type": "gmail_draft_creation",
"action": {
"tool": "create_gmail_draft",
"params": {
"to": to,
"subject": subject,
"body": body,
"cc": cc,
"bcc": bcc,
"connector_id": None,
},
},
"context": context,
}
result = request_approval(
action_type="gmail_draft_creation",
tool_name="create_gmail_draft",
params={
"to": to,
"subject": subject,
"body": body,
"cc": cc,
"bcc": bcc,
"connector_id": None,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The draft was not created. Do not ask again or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_to = final_params.get("to", to)
final_subject = final_params.get("subject", subject)
final_body = final_params.get("body", body)
final_cc = final_params.get("cc", cc)
final_bcc = final_params.get("bcc", bcc)
final_connector_id = final_params.get("connector_id")
final_to = result.params.get("to", to)
final_subject = result.params.get("subject", subject)
final_body = result.params.get("body", body)
final_cc = result.params.get("cc", cc)
final_bcc = result.params.get("bcc", bcc)
final_connector_id = result.params.get("connector_id")
from sqlalchemy.future import select

68
surfsense_backend/app/agents/new_chat/tools/gmail/send_email.py
View file

@ -6,7 +6,7 @@ from email.mime.text import MIMEText
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.gmail import GmailToolMetadataService
@ -86,60 +86,32 @@ def create_send_gmail_email_tool(
logger.info(
f"Requesting approval for sending Gmail email: to='{to}', subject='{subject}'"
)
approval = interrupt(
{
"type": "gmail_email_send",
"action": {
"tool": "send_gmail_email",
"params": {
"to": to,
"subject": subject,
"body": body,
"cc": cc,
"bcc": bcc,
"connector_id": None,
},
},
"context": context,
}
result = request_approval(
action_type="gmail_email_send",
tool_name="send_gmail_email",
params={
"to": to,
"subject": subject,
"body": body,
"cc": cc,
"bcc": bcc,
"connector_id": None,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The email was not sent. Do not ask again or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_to = final_params.get("to", to)
final_subject = final_params.get("subject", subject)
final_body = final_params.get("body", body)
final_cc = final_params.get("cc", cc)
final_bcc = final_params.get("bcc", bcc)
final_connector_id = final_params.get("connector_id")
final_to = result.params.get("to", to)
final_subject = result.params.get("subject", subject)
final_body = result.params.get("body", body)
final_cc = result.params.get("cc", cc)
final_bcc = result.params.get("bcc", bcc)
final_connector_id = result.params.get("connector_id")
from sqlalchemy.future import select

56
surfsense_backend/app/agents/new_chat/tools/gmail/trash_email.py
View file

@ -4,7 +4,7 @@ from datetime import datetime
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.gmail import GmailToolMetadataService
@ -101,56 +101,28 @@ def create_trash_gmail_email_tool(
logger.info(
f"Requesting approval for trashing Gmail email: '{email_subject_or_id}' (message_id={message_id}, delete_from_kb={delete_from_kb})"
)
approval = interrupt(
{
"type": "gmail_email_trash",
"action": {
"tool": "trash_gmail_email",
"params": {
"message_id": message_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
},
"context": context,
}
result = request_approval(
action_type="gmail_email_trash",
tool_name="trash_gmail_email",
params={
"message_id": message_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The email was not trashed. Do not ask again or suggest alternatives.",
}
edited_action = decision.get("edited_action")
final_params: dict[str, Any] = {}
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_message_id = final_params.get("message_id", message_id)
final_connector_id = final_params.get(
final_message_id = result.params.get("message_id", message_id)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_delete_from_kb = final_params.get("delete_from_kb", delete_from_kb)
final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
if not final_connector_id:
return {

74
surfsense_backend/app/agents/new_chat/tools/gmail/update_draft.py
View file

@ -6,7 +6,7 @@ from email.mime.text import MIMEText
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.gmail import GmailToolMetadataService
@ -122,65 +122,37 @@ def create_update_gmail_draft_tool(
f"Requesting approval for updating Gmail draft: '{original_subject}' "
f"(message_id={message_id}, draft_id={draft_id_from_context})"
)
approval = interrupt(
{
"type": "gmail_draft_update",
"action": {
"tool": "update_gmail_draft",
"params": {
"message_id": message_id,
"draft_id": draft_id_from_context,
"to": final_to_default,
"subject": final_subject_default,
"body": body,
"cc": cc,
"bcc": bcc,
"connector_id": connector_id_from_context,
},
},
"context": context,
}
result = request_approval(
action_type="gmail_draft_update",
tool_name="update_gmail_draft",
params={
"message_id": message_id,
"draft_id": draft_id_from_context,
"to": final_to_default,
"subject": final_subject_default,
"body": body,
"cc": cc,
"bcc": bcc,
"connector_id": connector_id_from_context,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The draft was not updated. Do not ask again or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_to = final_params.get("to", final_to_default)
final_subject = final_params.get("subject", final_subject_default)
final_body = final_params.get("body", body)
final_cc = final_params.get("cc", cc)
final_bcc = final_params.get("bcc", bcc)
final_connector_id = final_params.get(
final_to = result.params.get("to", final_to_default)
final_subject = result.params.get("subject", final_subject_default)
final_body = result.params.get("body", body)
final_cc = result.params.get("cc", cc)
final_bcc = result.params.get("bcc", bcc)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_draft_id = final_params.get("draft_id", draft_id_from_context)
final_draft_id = result.params.get("draft_id", draft_id_from_context)
if not final_connector_id:
return {

74
surfsense_backend/app/agents/new_chat/tools/google_calendar/create_event.py
View file

@ -6,9 +6,9 @@ from typing import Any
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from langchain_core.tools import tool
from langgraph.types import interrupt
from sqlalchemy.ext.asyncio import AsyncSession
from app.agents.new_chat.tools.hitl import request_approval
from app.services.google_calendar import GoogleCalendarToolMetadataService
logger = logging.getLogger(__name__)
@ -90,63 +90,35 @@ def create_create_calendar_event_tool(
logger.info(
f"Requesting approval for creating calendar event: summary='{summary}'"
)
approval = interrupt(
{
"type": "google_calendar_event_creation",
"action": {
"tool": "create_calendar_event",
"params": {
"summary": summary,
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"description": description,
"location": location,
"attendees": attendees,
"timezone": context.get("timezone"),
"connector_id": None,
},
},
"context": context,
}
result = request_approval(
action_type="google_calendar_event_creation",
tool_name="create_calendar_event",
params={
"summary": summary,
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"description": description,
"location": location,
"attendees": attendees,
"timezone": context.get("timezone"),
"connector_id": None,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The event was not created. Do not ask again or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_summary = final_params.get("summary", summary)
final_start_datetime = final_params.get("start_datetime", start_datetime)
final_end_datetime = final_params.get("end_datetime", end_datetime)
final_description = final_params.get("description", description)
final_location = final_params.get("location", location)
final_attendees = final_params.get("attendees", attendees)
final_connector_id = final_params.get("connector_id")
final_summary = result.params.get("summary", summary)
final_start_datetime = result.params.get("start_datetime", start_datetime)
final_end_datetime = result.params.get("end_datetime", end_datetime)
final_description = result.params.get("description", description)
final_location = result.params.get("location", location)
final_attendees = result.params.get("attendees", attendees)
final_connector_id = result.params.get("connector_id")
if not final_summary or not final_summary.strip():
return {"status": "error", "message": "Event summary cannot be empty."}

56
surfsense_backend/app/agents/new_chat/tools/google_calendar/delete_event.py
View file

@ -6,9 +6,9 @@ from typing import Any
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from langchain_core.tools import tool
from langgraph.types import interrupt
from sqlalchemy.ext.asyncio import AsyncSession
from app.agents.new_chat.tools.hitl import request_approval
from app.services.google_calendar import GoogleCalendarToolMetadataService
logger = logging.getLogger(__name__)
@ -100,56 +100,28 @@ def create_delete_calendar_event_tool(
logger.info(
f"Requesting approval for deleting calendar event: '{event_title_or_id}' (event_id={event_id}, delete_from_kb={delete_from_kb})"
)
approval = interrupt(
{
"type": "google_calendar_event_deletion",
"action": {
"tool": "delete_calendar_event",
"params": {
"event_id": event_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
},
"context": context,
}
result = request_approval(
action_type="google_calendar_event_deletion",
tool_name="delete_calendar_event",
params={
"event_id": event_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The event was not deleted. Do not ask again or suggest alternatives.",
}
edited_action = decision.get("edited_action")
final_params: dict[str, Any] = {}
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_event_id = final_params.get("event_id", event_id)
final_connector_id = final_params.get(
final_event_id = result.params.get("event_id", event_id)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_delete_from_kb = final_params.get("delete_from_kb", delete_from_kb)
final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
if not final_connector_id:
return {

78
surfsense_backend/app/agents/new_chat/tools/google_calendar/update_event.py
View file

@ -6,9 +6,9 @@ from typing import Any
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from langchain_core.tools import tool
from langgraph.types import interrupt
from sqlalchemy.ext.asyncio import AsyncSession
from app.agents.new_chat.tools.hitl import request_approval
from app.services.google_calendar import GoogleCalendarToolMetadataService
logger = logging.getLogger(__name__)
@ -116,71 +116,43 @@ def create_update_calendar_event_tool(
logger.info(
f"Requesting approval for updating calendar event: '{event_title_or_id}' (event_id={event_id})"
)
approval = interrupt(
{
"type": "google_calendar_event_update",
"action": {
"tool": "update_calendar_event",
"params": {
"event_id": event_id,
"document_id": document_id,
"connector_id": connector_id_from_context,
"new_summary": new_summary,
"new_start_datetime": new_start_datetime,
"new_end_datetime": new_end_datetime,
"new_description": new_description,
"new_location": new_location,
"new_attendees": new_attendees,
},
},
"context": context,
}
result = request_approval(
action_type="google_calendar_event_update",
tool_name="update_calendar_event",
params={
"event_id": event_id,
"document_id": document_id,
"connector_id": connector_id_from_context,
"new_summary": new_summary,
"new_start_datetime": new_start_datetime,
"new_end_datetime": new_end_datetime,
"new_description": new_description,
"new_location": new_location,
"new_attendees": new_attendees,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The event was not updated. Do not ask again or suggest alternatives.",
}
edited_action = decision.get("edited_action")
final_params: dict[str, Any] = {}
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_event_id = final_params.get("event_id", event_id)
final_connector_id = final_params.get(
final_event_id = result.params.get("event_id", event_id)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_new_summary = final_params.get("new_summary", new_summary)
final_new_start_datetime = final_params.get(
final_new_summary = result.params.get("new_summary", new_summary)
final_new_start_datetime = result.params.get(
"new_start_datetime", new_start_datetime
)
final_new_end_datetime = final_params.get(
final_new_end_datetime = result.params.get(
"new_end_datetime", new_end_datetime
)
final_new_description = final_params.get("new_description", new_description)
final_new_location = final_params.get("new_location", new_location)
final_new_attendees = final_params.get("new_attendees", new_attendees)
final_new_description = result.params.get("new_description", new_description)
final_new_location = result.params.get("new_location", new_location)
final_new_attendees = result.params.get("new_attendees", new_attendees)
if not final_connector_id:
return {

64
surfsense_backend/app/agents/new_chat/tools/google_drive/create_file.py
View file

@ -3,9 +3,9 @@ from typing import Any, Literal
from googleapiclient.errors import HttpError
from langchain_core.tools import tool
from langgraph.types import interrupt
from sqlalchemy.ext.asyncio import AsyncSession
from app.agents.new_chat.tools.hitl import request_approval
from app.connectors.google_drive.client import GoogleDriveClient
from app.connectors.google_drive.file_types import GOOGLE_DOC, GOOGLE_SHEET
from app.services.google_drive import GoogleDriveToolMetadataService
@ -99,58 +99,30 @@ def create_create_google_drive_file_tool(
logger.info(
f"Requesting approval for creating Google Drive file: name='{name}', type='{file_type}'"
)
approval = interrupt(
{
"type": "google_drive_file_creation",
"action": {
"tool": "create_google_drive_file",
"params": {
"name": name,
"file_type": file_type,
"content": content,
"connector_id": None,
"parent_folder_id": None,
},
},
"context": context,
}
result = request_approval(
action_type="google_drive_file_creation",
tool_name="create_google_drive_file",
params={
"name": name,
"file_type": file_type,
"content": content,
"connector_id": None,
"parent_folder_id": None,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The file was not created. Do not ask again or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_name = final_params.get("name", name)
final_file_type = final_params.get("file_type", file_type)
final_content = final_params.get("content", content)
final_connector_id = final_params.get("connector_id")
final_parent_folder_id = final_params.get("parent_folder_id")
final_name = result.params.get("name", name)
final_file_type = result.params.get("file_type", file_type)
final_content = result.params.get("content", content)
final_connector_id = result.params.get("connector_id")
final_parent_folder_id = result.params.get("parent_folder_id")
if not final_name or not final_name.strip():
return {"status": "error", "message": "File name cannot be empty."}

56
surfsense_backend/app/agents/new_chat/tools/google_drive/trash_file.py
View file

@ -3,9 +3,9 @@ from typing import Any
from googleapiclient.errors import HttpError
from langchain_core.tools import tool
from langgraph.types import interrupt
from sqlalchemy.ext.asyncio import AsyncSession
from app.agents.new_chat.tools.hitl import request_approval
from app.connectors.google_drive.client import GoogleDriveClient
from app.services.google_drive import GoogleDriveToolMetadataService
@ -101,56 +101,28 @@ def create_delete_google_drive_file_tool(
logger.info(
f"Requesting approval for deleting Google Drive file: '{file_name}' (file_id={file_id}, delete_from_kb={delete_from_kb})"
)
approval = interrupt(
{
"type": "google_drive_file_trash",
"action": {
"tool": "delete_google_drive_file",
"params": {
"file_id": file_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
},
"context": context,
}
result = request_approval(
action_type="google_drive_file_trash",
tool_name="delete_google_drive_file",
params={
"file_id": file_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The file was not trashed. Do not ask again or suggest alternatives.",
}
edited_action = decision.get("edited_action")
final_params: dict[str, Any] = {}
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_file_id = final_params.get("file_id", file_id)
final_connector_id = final_params.get(
final_file_id = result.params.get("file_id", file_id)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_delete_from_kb = final_params.get("delete_from_kb", delete_from_kb)
final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
if not final_connector_id:
return {

140
surfsense_backend/app/agents/new_chat/tools/hitl.py Normal file
View file

@ -0,0 +1,140 @@
"""Unified HITL (Human-in-the-Loop) approval utility.
Provides a single ``request_approval()`` function that encapsulates the
interrupt payload creation, decision parsing, and parameter merging logic
shared by every sensitive tool (native connectors and MCP tools alike).
Usage inside a tool::
from app.agents.new_chat.tools.hitl import request_approval
result = request_approval(
action_type="gmail_email_send",
tool_name="send_gmail_email",
params={"to": to, "subject": subject, "body": body},
context=context,
)
if result.rejected:
return {"status": "rejected", "message": "User declined."}
# result.params contains the final (possibly edited) parameters
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from typing import Any
from langgraph.types import interrupt
logger = logging.getLogger(__name__)
@dataclass(frozen=True, slots=True)
class HITLResult:
"""Outcome of a human-in-the-loop approval request."""
rejected: bool
decision_type: str
params: dict[str, Any] = field(default_factory=dict)
def _parse_decision(approval: Any) -> tuple[str, dict[str, Any]]:
"""Extract the first valid decision and its edited parameters.
Returns:
(decision_type, edited_params) where *decision_type* is one of
``"approve"``, ``"edit"``, or ``"reject"`` and *edited_params* is
the dict of user-modified arguments (empty when there are none).
Raises:
ValueError: when no usable decision dict can be found.
"""
decisions_raw = approval.get("decisions", []) if isinstance(approval, dict) else []
decisions = decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
raise ValueError("No approval decision received")
decision = decisions[0]
decision_type: str = decision.get("type") or decision.get("decision_type") or "approve"
edited_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
edited_params = edited_args
elif isinstance(decision.get("args"), dict):
edited_params = decision["args"]
return decision_type, edited_params
def request_approval(
*,
action_type: str,
tool_name: str,
params: dict[str, Any],
context: dict[str, Any] | None = None,
trusted_tools: list[str] | None = None,
) -> HITLResult:
"""Pause the graph for user approval and return the decision.
This is a **synchronous** helper (not ``async``) because
``langgraph.types.interrupt`` is itself synchronous — it raises a
``GraphInterrupt`` exception that the LangGraph runtime catches.
Parameters
----------
action_type:
A label that the frontend uses to select the correct approval card
(e.g. ``"gmail_email_send"``, ``"mcp_tool_call"``).
tool_name:
The registered LangChain tool name (e.g. ``"send_gmail_email"``).
params:
The original tool arguments. These are shown in the approval card
and used as defaults when the user does not edit anything.
context:
Rich metadata from a ``*ToolMetadataService`` (accounts, folders,
labels, etc.). For MCP tools this can hold the server name and
tool description.
trusted_tools:
An allow-list of tool names the user has previously marked as
"Always Allow". If *tool_name* appears in this list, HITL is
skipped and the tool executes immediately.
Returns
-------
HITLResult
``result.rejected`` is ``True`` when the user chose to deny the
action. Otherwise ``result.params`` contains the final parameter
dict — either the originals or the user-edited version merged on
top.
"""
if trusted_tools and tool_name in trusted_tools:
logger.info("Tool '%s' is user-trusted — skipping HITL", tool_name)
return HITLResult(rejected=False, decision_type="trusted", params=dict(params))
approval = interrupt(
{
"type": action_type,
"action": {"tool": tool_name, "params": params},
"context": context or {},
}
)
try:
decision_type, edited_params = _parse_decision(approval)
except ValueError:
logger.warning("No approval decision received for %s", tool_name)
return HITLResult(rejected=False, decision_type="error", params=params)
logger.info("User decision for %s: %s", tool_name, decision_type)
if decision_type == "reject":
return HITLResult(rejected=True, decision_type="reject", params=params)
final_params = {**params, **edited_params} if edited_params else dict(params)
return HITLResult(rejected=False, decision_type=decision_type, params=final_params)

68
surfsense_backend/app/agents/new_chat/tools/jira/create_issue.py
View file

@ -3,7 +3,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified
@ -69,58 +69,32 @@ def create_create_jira_issue_tool(
"connector_type": "jira",
}
approval = interrupt(
{
"type": "jira_issue_creation",
"action": {
"tool": "create_jira_issue",
"params": {
"project_key": project_key,
"summary": summary,
"issue_type": issue_type,
"description": description,
"priority": priority,
"connector_id": connector_id,
},
},
"context": context,
}
result = request_approval(
action_type="jira_issue_creation",
tool_name="create_jira_issue",
params={
"project_key": project_key,
"summary": summary,
"issue_type": issue_type,
"description": description,
"priority": priority,
"connector_id": connector_id,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The issue was not created.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_project_key = final_params.get("project_key", project_key)
final_summary = final_params.get("summary", summary)
final_issue_type = final_params.get("issue_type", issue_type)
final_description = final_params.get("description", description)
final_priority = final_params.get("priority", priority)
final_connector_id = final_params.get("connector_id", connector_id)
final_project_key = result.params.get("project_key", project_key)
final_summary = result.params.get("summary", summary)
final_issue_type = result.params.get("issue_type", issue_type)
final_description = result.params.get("description", description)
final_priority = result.params.get("priority", priority)
final_connector_id = result.params.get("connector_id", connector_id)
if not final_summary or not final_summary.strip():
return {"status": "error", "message": "Issue summary cannot be empty."}

56
surfsense_backend/app/agents/new_chat/tools/jira/delete_issue.py
View file

@ -3,7 +3,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified
@ -71,54 +71,28 @@ def create_delete_jira_issue_tool(
document_id = issue_data["document_id"]
connector_id_from_context = context.get("account", {}).get("id")
approval = interrupt(
{
"type": "jira_issue_deletion",
"action": {
"tool": "delete_jira_issue",
"params": {
"issue_key": issue_key,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
},
"context": context,
}
result = request_approval(
action_type="jira_issue_deletion",
tool_name="delete_jira_issue",
params={
"issue_key": issue_key,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The issue was not deleted.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_issue_key = final_params.get("issue_key", issue_key)
final_connector_id = final_params.get(
final_issue_key = result.params.get("issue_key", issue_key)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_delete_from_kb = final_params.get("delete_from_kb", delete_from_kb)
final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
from sqlalchemy.future import select

68
surfsense_backend/app/agents/new_chat/tools/jira/update_issue.py
View file

@ -3,7 +3,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified
@ -75,60 +75,34 @@ def create_update_jira_issue_tool(
document_id = issue_data.get("document_id")
connector_id_from_context = context.get("account", {}).get("id")
approval = interrupt(
{
"type": "jira_issue_update",
"action": {
"tool": "update_jira_issue",
"params": {
"issue_key": issue_key,
"document_id": document_id,
"new_summary": new_summary,
"new_description": new_description,
"new_priority": new_priority,
"connector_id": connector_id_from_context,
},
},
"context": context,
}
result = request_approval(
action_type="jira_issue_update",
tool_name="update_jira_issue",
params={
"issue_key": issue_key,
"document_id": document_id,
"new_summary": new_summary,
"new_description": new_description,
"new_priority": new_priority,
"connector_id": connector_id_from_context,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The issue was not updated.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_issue_key = final_params.get("issue_key", issue_key)
final_summary = final_params.get("new_summary", new_summary)
final_description = final_params.get("new_description", new_description)
final_priority = final_params.get("new_priority", new_priority)
final_connector_id = final_params.get(
final_issue_key = result.params.get("issue_key", issue_key)
final_summary = result.params.get("new_summary", new_summary)
final_description = result.params.get("new_description", new_description)
final_priority = result.params.get("new_priority", new_priority)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_document_id = final_params.get("document_id", document_id)
final_document_id = result.params.get("document_id", document_id)
from sqlalchemy.future import select

78
surfsense_backend/app/agents/new_chat/tools/linear/create_issue.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from app.connectors.linear_connector import LinearAPIError, LinearConnector
@ -94,65 +94,37 @@ def create_create_linear_issue_tool(
}
logger.info(f"Requesting approval for creating Linear issue: '{title}'")
approval = interrupt(
{
"type": "linear_issue_creation",
"action": {
"tool": "create_linear_issue",
"params": {
"title": title,
"description": description,
"team_id": None,
"state_id": None,
"assignee_id": None,
"priority": None,
"label_ids": [],
"connector_id": connector_id,
},
},
"context": context,
}
result = request_approval(
action_type="linear_issue_creation",
tool_name="create_linear_issue",
params={
"title": title,
"description": description,
"team_id": None,
"state_id": None,
"assignee_id": None,
"priority": None,
"label_ids": [],
"connector_id": connector_id,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
logger.info("Linear issue creation rejected by user")
return {
"status": "rejected",
"message": "User declined. The issue was not created. Do not ask again or suggest alternatives.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_title = final_params.get("title", title)
final_description = final_params.get("description", description)
final_team_id = final_params.get("team_id")
final_state_id = final_params.get("state_id")
final_assignee_id = final_params.get("assignee_id")
final_priority = final_params.get("priority")
final_label_ids = final_params.get("label_ids") or []
final_connector_id = final_params.get("connector_id", connector_id)
final_title = result.params.get("title", title)
final_description = result.params.get("description", description)
final_team_id = result.params.get("team_id")
final_state_id = result.params.get("state_id")
final_assignee_id = result.params.get("assignee_id")
final_priority = result.params.get("priority")
final_label_ids = result.params.get("label_ids") or []
final_connector_id = result.params.get("connector_id", connector_id)
if not final_title or not final_title.strip():
logger.error("Title is empty or contains only whitespace")

58
surfsense_backend/app/agents/new_chat/tools/linear/delete_issue.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from app.connectors.linear_connector import LinearAPIError, LinearConnector
@ -114,57 +114,29 @@ def create_delete_linear_issue_tool(
f"Requesting approval for deleting Linear issue: '{issue_ref}' "
f"(id={issue_id}, delete_from_kb={delete_from_kb})"
)
approval = interrupt(
{
"type": "linear_issue_deletion",
"action": {
"tool": "delete_linear_issue",
"params": {
"issue_id": issue_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
},
"context": context,
}
result = request_approval(
action_type="linear_issue_deletion",
tool_name="delete_linear_issue",
params={
"issue_id": issue_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
logger.info("Linear issue deletion rejected by user")
return {
"status": "rejected",
"message": "User declined. The issue was not deleted. Do not ask again or suggest alternatives.",
"message": "User declined. Do not retry or suggest alternatives.",
}
edited_action = decision.get("edited_action")
final_params: dict[str, Any] = {}
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_issue_id = final_params.get("issue_id", issue_id)
final_connector_id = final_params.get(
final_issue_id = result.params.get("issue_id", issue_id)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_delete_from_kb = final_params.get("delete_from_kb", delete_from_kb)
final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
logger.info(
f"Deleting Linear issue with final params: issue_id={final_issue_id}, "

82
surfsense_backend/app/agents/new_chat/tools/linear/update_issue.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from app.connectors.linear_connector import LinearAPIError, LinearConnector
@ -130,69 +130,41 @@ def create_update_linear_issue_tool(
logger.info(
f"Requesting approval for updating Linear issue: '{issue_ref}' (id={issue_id})"
)
approval = interrupt(
{
"type": "linear_issue_update",
"action": {
"tool": "update_linear_issue",
"params": {
"issue_id": issue_id,
"document_id": document_id,
"new_title": new_title,
"new_description": new_description,
"new_state_id": new_state_id,
"new_assignee_id": new_assignee_id,
"new_priority": new_priority,
"new_label_ids": new_label_ids,
"connector_id": connector_id_from_context,
},
},
"context": context,
}
result = request_approval(
action_type="linear_issue_update",
tool_name="update_linear_issue",
params={
"issue_id": issue_id,
"document_id": document_id,
"new_title": new_title,
"new_description": new_description,
"new_state_id": new_state_id,
"new_assignee_id": new_assignee_id,
"new_priority": new_priority,
"new_label_ids": new_label_ids,
"connector_id": connector_id_from_context,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
logger.info("Linear issue update rejected by user")
return {
"status": "rejected",
"message": "User declined. The issue was not updated. Do not ask again or suggest alternatives.",
"message": "User declined. Do not retry or suggest alternatives.",
}
edited_action = decision.get("edited_action")
final_params: dict[str, Any] = {}
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_issue_id = final_params.get("issue_id", issue_id)
final_document_id = final_params.get("document_id", document_id)
final_new_title = final_params.get("new_title", new_title)
final_new_description = final_params.get("new_description", new_description)
final_new_state_id = final_params.get("new_state_id", new_state_id)
final_new_assignee_id = final_params.get("new_assignee_id", new_assignee_id)
final_new_priority = final_params.get("new_priority", new_priority)
final_new_label_ids: list[str] | None = final_params.get(
final_issue_id = result.params.get("issue_id", issue_id)
final_document_id = result.params.get("document_id", document_id)
final_new_title = result.params.get("new_title", new_title)
final_new_description = result.params.get("new_description", new_description)
final_new_state_id = result.params.get("new_state_id", new_state_id)
final_new_assignee_id = result.params.get("new_assignee_id", new_assignee_id)
final_new_priority = result.params.get("new_priority", new_priority)
final_new_label_ids: list[str] | None = result.params.get(
"new_label_ids", new_label_ids
)
final_connector_id = final_params.get(
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)

180
surfsense_backend/app/agents/new_chat/tools/mcp_tool.py
View file

@ -7,7 +7,11 @@ Supports both transport types:
- stdio: Local process-based MCP servers (command, args, env)
- streamable-http/http/sse: Remote HTTP-based MCP servers (url, headers)
This implements real MCP protocol support similar to Cursor's implementation.
All MCP tools are unconditionally gated by HITL (Human-in-the-Loop) approval.
Per the MCP spec: "Clients MUST consider tool annotations to be untrusted unless
they come from trusted servers." Users can bypass HITL for specific tools by
clicking "Always Allow", which adds the tool name to the connector's
``config.trusted_tools`` allow-list.
"""
import logging
@ -21,6 +25,7 @@ from pydantic import BaseModel, create_model
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.agents.new_chat.tools.hitl import request_approval
from app.agents.new_chat.tools.mcp_client import MCPClient
from app.db import SearchSourceConnector, SearchSourceConnectorType
@ -49,27 +54,15 @@ def _create_dynamic_input_model_from_schema(
tool_name: str,
input_schema: dict[str, Any],
) -> type[BaseModel]:
"""Create a Pydantic model from MCP tool's JSON schema.
Args:
tool_name: Name of the tool (used for model class name)
input_schema: JSON schema from MCP server
Returns:
Pydantic model class for tool input validation
"""
"""Create a Pydantic model from MCP tool's JSON schema."""
properties = input_schema.get("properties", {})
required_fields = input_schema.get("required", [])
# Build Pydantic field definitions
field_definitions = {}
for param_name, param_schema in properties.items():
param_description = param_schema.get("description", "")
is_required = param_name in required_fields
# Use Any type for complex schemas to preserve structure
# This allows the MCP server to do its own validation
from typing import Any as AnyType
from pydantic import Field
@ -85,7 +78,6 @@ def _create_dynamic_input_model_from_schema(
Field(None, description=param_description),
)
# Create dynamic model
model_name = f"{tool_name.replace(' ', '').replace('-', '_')}Input"
return create_model(model_name, **field_definitions)
@ -93,55 +85,70 @@ def _create_dynamic_input_model_from_schema(
async def _create_mcp_tool_from_definition_stdio(
tool_def: dict[str, Any],
mcp_client: MCPClient,
*,
connector_name: str = "",
connector_id: int | None = None,
trusted_tools: list[str] | None = None,
) -> StructuredTool:
"""Create a LangChain tool from an MCP tool definition (stdio transport).
Args:
tool_def: Tool definition from MCP server with name, description, input_schema
mcp_client: MCP client instance for calling the tool
Returns:
LangChain StructuredTool instance
All MCP tools are unconditionally wrapped with HITL approval.
``request_approval()`` is called OUTSIDE the try/except so that
``GraphInterrupt`` propagates cleanly to LangGraph.
"""
tool_name = tool_def.get("name", "unnamed_tool")
tool_description = tool_def.get("description", "No description provided")
input_schema = tool_def.get("input_schema", {"type": "object", "properties": {}})
# Log the actual schema for debugging
logger.info(f"MCP tool '{tool_name}' input schema: {input_schema}")
# Create dynamic input model from schema
input_model = _create_dynamic_input_model_from_schema(tool_name, input_schema)
async def mcp_tool_call(**kwargs) -> str:
"""Execute the MCP tool call via the client with retry support."""
logger.info(f"MCP tool '{tool_name}' called with params: {kwargs}")
# HITL — OUTSIDE try/except so GraphInterrupt propagates to LangGraph
hitl_result = request_approval(
action_type="mcp_tool_call",
tool_name=tool_name,
params=kwargs,
context={
"mcp_server": connector_name,
"tool_description": tool_description,
"mcp_transport": "stdio",
"mcp_connector_id": connector_id,
},
trusted_tools=trusted_tools,
)
if hitl_result.rejected:
return "Tool call rejected by user."
call_kwargs = hitl_result.params
try:
# Connect to server and call tool (connect has built-in retry logic)
async with mcp_client.connect():
result = await mcp_client.call_tool(tool_name, kwargs)
result = await mcp_client.call_tool(tool_name, call_kwargs)
return str(result)
except RuntimeError as e:
# Connection failures after all retries
error_msg = f"MCP tool '{tool_name}' connection failed after retries: {e!s}"
logger.error(error_msg)
return f"Error: {error_msg}"
except Exception as e:
# Tool execution or other errors
error_msg = f"MCP tool '{tool_name}' execution failed: {e!s}"
logger.exception(error_msg)
return f"Error: {error_msg}"
# Create StructuredTool with response_format to preserve exact schema
tool = StructuredTool(
name=tool_name,
description=tool_description,
coroutine=mcp_tool_call,
args_schema=input_model,
# Store the original MCP schema as metadata so we can access it later
metadata={"mcp_input_schema": input_schema, "mcp_transport": "stdio"},
metadata={
"mcp_input_schema": input_schema,
"mcp_transport": "stdio",
"hitl": True,
"hitl_dedup_key": next(iter(input_schema.get("required", [])), None),
},
)
logger.info(f"Created MCP tool (stdio): '{tool_name}'")
@ -152,43 +159,54 @@ async def _create_mcp_tool_from_definition_http(
tool_def: dict[str, Any],
url: str,
headers: dict[str, str],
*,
connector_name: str = "",
connector_id: int | None = None,
trusted_tools: list[str] | None = None,
) -> StructuredTool:
"""Create a LangChain tool from an MCP tool definition (HTTP transport).
Args:
tool_def: Tool definition from MCP server with name, description, input_schema
url: URL of the MCP server
headers: HTTP headers for authentication
Returns:
LangChain StructuredTool instance
All MCP tools are unconditionally wrapped with HITL approval.
``request_approval()`` is called OUTSIDE the try/except so that
``GraphInterrupt`` propagates cleanly to LangGraph.
"""
tool_name = tool_def.get("name", "unnamed_tool")
tool_description = tool_def.get("description", "No description provided")
input_schema = tool_def.get("input_schema", {"type": "object", "properties": {}})
# Log the actual schema for debugging
logger.info(f"MCP HTTP tool '{tool_name}' input schema: {input_schema}")
# Create dynamic input model from schema
input_model = _create_dynamic_input_model_from_schema(tool_name, input_schema)
async def mcp_http_tool_call(**kwargs) -> str:
"""Execute the MCP tool call via HTTP transport."""
logger.info(f"MCP HTTP tool '{tool_name}' called with params: {kwargs}")
# HITL — OUTSIDE try/except so GraphInterrupt propagates to LangGraph
hitl_result = request_approval(
action_type="mcp_tool_call",
tool_name=tool_name,
params=kwargs,
context={
"mcp_server": connector_name,
"tool_description": tool_description,
"mcp_transport": "http",
"mcp_connector_id": connector_id,
},
trusted_tools=trusted_tools,
)
if hitl_result.rejected:
return "Tool call rejected by user."
call_kwargs = hitl_result.params
try:
async with (
streamablehttp_client(url, headers=headers) as (read, write, _),
ClientSession(read, write) as session,
):
await session.initialize()
response = await session.call_tool(tool_name, arguments=call_kwargs)
# Call the tool
response = await session.call_tool(tool_name, arguments=kwargs)
# Extract content from response
result = []
for content in response.content:
if hasattr(content, "text"):
@ -209,7 +227,6 @@ async def _create_mcp_tool_from_definition_http(
logger.exception(error_msg)
return f"Error: {error_msg}"
# Create StructuredTool
tool = StructuredTool(
name=tool_name,
description=tool_description,
@ -219,6 +236,8 @@ async def _create_mcp_tool_from_definition_http(
"mcp_input_schema": input_schema,
"mcp_transport": "http",
"mcp_url": url,
"hitl": True,
"hitl_dedup_key": next(iter(input_schema.get("required", [])), None),
},
)
@ -230,20 +249,11 @@ async def _load_stdio_mcp_tools(
connector_id: int,
connector_name: str,
server_config: dict[str, Any],
trusted_tools: list[str] | None = None,
) -> list[StructuredTool]:
"""Load tools from a stdio-based MCP server.
Args:
connector_id: Connector ID for logging
connector_name: Connector name for logging
server_config: Server configuration with command, args, env
Returns:
List of tools from the MCP server
"""
"""Load tools from a stdio-based MCP server."""
tools: list[StructuredTool] = []
# Validate required command field
command = server_config.get("command")
if not command or not isinstance(command, str):
logger.warning(
@ -251,7 +261,6 @@ async def _load_stdio_mcp_tools(
)
return tools
# Validate args field (must be list if present)
args = server_config.get("args", [])
if not isinstance(args, list):
logger.warning(
@ -259,7 +268,6 @@ async def _load_stdio_mcp_tools(
)
return tools
# Validate env field (must be dict if present)
env = server_config.get("env", {})
if not isinstance(env, dict):
logger.warning(
@ -267,10 +275,8 @@ async def _load_stdio_mcp_tools(
)
return tools
# Create MCP client
mcp_client = MCPClient(command, args, env)
# Connect and discover tools
async with mcp_client.connect():
tool_definitions = await mcp_client.list_tools()
@ -279,10 +285,15 @@ async def _load_stdio_mcp_tools(
f"'{command}' (connector {connector_id})"
)
# Create LangChain tools from definitions
for tool_def in tool_definitions:
try:
tool = await _create_mcp_tool_from_definition_stdio(tool_def, mcp_client)
tool = await _create_mcp_tool_from_definition_stdio(
tool_def,
mcp_client,
connector_name=connector_name,
connector_id=connector_id,
trusted_tools=trusted_tools,
)
tools.append(tool)
except Exception as e:
logger.exception(
@ -297,20 +308,11 @@ async def _load_http_mcp_tools(
connector_id: int,
connector_name: str,
server_config: dict[str, Any],
trusted_tools: list[str] | None = None,
) -> list[StructuredTool]:
"""Load tools from an HTTP-based MCP server.
Args:
connector_id: Connector ID for logging
connector_name: Connector name for logging
server_config: Server configuration with url, headers
Returns:
List of tools from the MCP server
"""
"""Load tools from an HTTP-based MCP server."""
tools: list[StructuredTool] = []
# Validate required url field
url = server_config.get("url")
if not url or not isinstance(url, str):
logger.warning(
@ -318,7 +320,6 @@ async def _load_http_mcp_tools(
)
return tools
# Validate headers field (must be dict if present)
headers = server_config.get("headers", {})
if not isinstance(headers, dict):
logger.warning(
@ -326,7 +327,6 @@ async def _load_http_mcp_tools(
)
return tools
# Connect and discover tools via HTTP
try:
async with (
streamablehttp_client(url, headers=headers) as (read, write, _),
@ -334,7 +334,6 @@ async def _load_http_mcp_tools(
):
await session.initialize()
# List available tools
response = await session.list_tools()
tool_definitions = []
for tool in response.tools:
@ -353,11 +352,15 @@ async def _load_http_mcp_tools(
f"'{url}' (connector {connector_id})"
)
# Create LangChain tools from definitions
for tool_def in tool_definitions:
try:
tool = await _create_mcp_tool_from_definition_http(
tool_def, url, headers
tool_def,
url,
headers,
connector_name=connector_name,
connector_id=connector_id,
trusted_tools=trusted_tools,
)
tools.append(tool)
except Exception as e:
@ -398,14 +401,6 @@ async def load_mcp_tools(
Results are cached per search space for up to 5 minutes to avoid
re-spawning MCP server processes on every chat message.
Args:
session: Database session
search_space_id: User's search space ID
Returns:
List of LangChain StructuredTool instances
"""
_evict_expired_mcp_cache()
@ -436,6 +431,7 @@ async def load_mcp_tools(
try:
config = connector.config or {}
server_config = config.get("server_config", {})
trusted_tools = config.get("trusted_tools", [])
if not server_config or not isinstance(server_config, dict):
logger.warning(
@ -447,11 +443,17 @@ async def load_mcp_tools(
if transport in ("streamable-http", "http", "sse"):
connector_tools = await _load_http_mcp_tools(
connector.id, connector.name, server_config
connector.id,
connector.name,
server_config,
trusted_tools=trusted_tools,
)
else:
connector_tools = await _load_stdio_mcp_tools(
connector.id, connector.name, server_config
connector.id,
connector.name,
server_config,
trusted_tools=trusted_tools,
)
tools.extend(connector_tools)

66
surfsense_backend/app/agents/new_chat/tools/notion/create_page.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from app.connectors.notion_history import NotionAPIError, NotionHistoryConnector
@ -99,61 +99,29 @@ def create_create_notion_page_tool(
}
logger.info(f"Requesting approval for creating Notion page: '{title}'")
approval = interrupt(
{
"type": "notion_page_creation",
"action": {
"tool": "create_notion_page",
"params": {
"title": title,
"content": content,
"parent_page_id": None,
"connector_id": connector_id,
},
},
"context": context,
}
result = request_approval(
action_type="notion_page_creation",
tool_name="create_notion_page",
params={
"title": title,
"content": content,
"parent_page_id": None,
"connector_id": connector_id,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {
"status": "error",
"message": "No approval decision received",
}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
logger.info("Notion page creation rejected by user")
return {
"status": "rejected",
"message": "User declined. The page was not created. Do not ask again or suggest alternatives.",
"message": "User declined. Do not retry or suggest alternatives.",
}
edited_action = decision.get("edited_action")
final_params: dict[str, Any] = {}
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
# Some interrupt payloads place args directly on the decision.
final_params = decision["args"]
final_title = final_params.get("title", title)
final_content = final_params.get("content", content)
final_parent_page_id = final_params.get("parent_page_id")
final_connector_id = final_params.get("connector_id", connector_id)
final_title = result.params.get("title", title)
final_content = result.params.get("content", content)
final_parent_page_id = result.params.get("parent_page_id")
final_connector_id = result.params.get("connector_id", connector_id)
if not final_title or not final_title.strip():
logger.error("Title is empty or contains only whitespace")

64
surfsense_backend/app/agents/new_chat/tools/notion/delete_page.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from app.connectors.notion_history import NotionAPIError, NotionHistoryConnector
@ -114,63 +114,29 @@ def create_delete_notion_page_tool(
f"Requesting approval for deleting Notion page: '{page_title}' (page_id={page_id}, delete_from_kb={delete_from_kb})"
)
# Request approval before deleting
approval = interrupt(
{
"type": "notion_page_deletion",
"action": {
"tool": "delete_notion_page",
"params": {
"page_id": page_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
},
"context": context,
}
result = request_approval(
action_type="notion_page_deletion",
tool_name="delete_notion_page",
params={
"page_id": page_id,
"connector_id": connector_id_from_context,
"delete_from_kb": delete_from_kb,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {
"status": "error",
"message": "No approval decision received",
}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
logger.info("Notion page deletion rejected by user")
return {
"status": "rejected",
"message": "User declined. The page was not deleted. Do not ask again or suggest alternatives.",
"message": "User declined. Do not retry or suggest alternatives.",
}
# Extract edited action arguments (if user modified the checkbox)
edited_action = decision.get("edited_action")
final_params: dict[str, Any] = {}
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
# Some interrupt payloads place args directly on the decision.
final_params = decision["args"]
final_page_id = final_params.get("page_id", page_id)
final_connector_id = final_params.get(
final_page_id = result.params.get("page_id", page_id)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)
final_delete_from_kb = final_params.get("delete_from_kb", delete_from_kb)
final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
logger.info(
f"Deleting Notion page with final params: page_id={final_page_id}, connector_id={final_connector_id}, delete_from_kb={final_delete_from_kb}"

62
surfsense_backend/app/agents/new_chat/tools/notion/update_page.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from app.connectors.notion_history import NotionAPIError, NotionHistoryConnector
@ -127,59 +127,27 @@ def create_update_notion_page_tool(
logger.info(
f"Requesting approval for updating Notion page: '{page_title}' (page_id={page_id})"
)
approval = interrupt(
{
"type": "notion_page_update",
"action": {
"tool": "update_notion_page",
"params": {
"page_id": page_id,
"content": content,
"connector_id": connector_id_from_context,
},
},
"context": context,
}
result = request_approval(
action_type="notion_page_update",
tool_name="update_notion_page",
params={
"page_id": page_id,
"content": content,
"connector_id": connector_id_from_context,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
logger.warning("No approval decision received")
return {
"status": "error",
"message": "No approval decision received",
}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
logger.info("Notion page update rejected by user")
return {
"status": "rejected",
"message": "User declined. The page was not updated. Do not ask again or suggest alternatives.",
"message": "User declined. Do not retry or suggest alternatives.",
}
edited_action = decision.get("edited_action")
final_params: dict[str, Any] = {}
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
# Some interrupt payloads place args directly on the decision.
final_params = decision["args"]
final_page_id = final_params.get("page_id", page_id)
final_content = final_params.get("content", content)
final_connector_id = final_params.get(
final_page_id = result.params.get("page_id", page_id)
final_content = result.params.get("content", content)
final_connector_id = result.params.get(
"connector_id", connector_id_from_context
)

60
surfsense_backend/app/agents/new_chat/tools/onedrive/create_file.py
View file

@ -5,7 +5,7 @@ from pathlib import Path
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
@ -145,54 +145,28 @@ def create_create_onedrive_file_tool(
"parent_folders": parent_folders,
}
approval = interrupt(
{
"type": "onedrive_file_creation",
"action": {
"tool": "create_onedrive_file",
"params": {
"name": name,
"content": content,
"connector_id": None,
"parent_folder_id": None,
},
},
"context": context,
}
result = request_approval(
action_type="onedrive_file_creation",
tool_name="create_onedrive_file",
params={
"name": name,
"content": content,
"connector_id": None,
"parent_folder_id": None,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The file was not created.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_name = final_params.get("name", name)
final_content = final_params.get("content", content)
final_connector_id = final_params.get("connector_id")
final_parent_folder_id = final_params.get("parent_folder_id")
final_name = result.params.get("name", name)
final_content = result.params.get("content", content)
final_connector_id = result.params.get("connector_id")
final_parent_folder_id = result.params.get("parent_folder_id")
if not final_name or not final_name.strip():
return {"status": "error", "message": "File name cannot be empty."}

57
surfsense_backend/app/agents/new_chat/tools/onedrive/trash_file.py
View file

@ -2,7 +2,7 @@ import logging
from typing import Any
from langchain_core.tools import tool
from langgraph.types import interrupt
from app.agents.new_chat.tools.hitl import request_approval
from sqlalchemy import String, and_, cast, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
@ -174,53 +174,26 @@ def create_delete_onedrive_file_tool(
},
}
approval = interrupt(
{
"type": "onedrive_file_trash",
"action": {
"tool": "delete_onedrive_file",
"params": {
"file_id": file_id,
"connector_id": connector.id,
"delete_from_kb": delete_from_kb,
},
},
"context": context,
}
result = request_approval(
action_type="onedrive_file_trash",
tool_name="delete_onedrive_file",
params={
"file_id": file_id,
"connector_id": connector.id,
"delete_from_kb": delete_from_kb,
},
context=context,
)
decisions_raw = (
approval.get("decisions", []) if isinstance(approval, dict) else []
)
decisions = (
decisions_raw if isinstance(decisions_raw, list) else [decisions_raw]
)
decisions = [d for d in decisions if isinstance(d, dict)]
if not decisions:
return {"status": "error", "message": "No approval decision received"}
decision = decisions[0]
decision_type = decision.get("type") or decision.get("decision_type")
logger.info(f"User decision: {decision_type}")
if decision_type == "reject":
if result.rejected:
return {
"status": "rejected",
"message": "User declined. The file was not trashed. Do not ask again or suggest alternatives.",
"message": "User declined. Do not retry or suggest alternatives.",
}
final_params: dict[str, Any] = {}
edited_action = decision.get("edited_action")
if isinstance(edited_action, dict):
edited_args = edited_action.get("args")
if isinstance(edited_args, dict):
final_params = edited_args
elif isinstance(decision.get("args"), dict):
final_params = decision["args"]
final_file_id = final_params.get("file_id", file_id)
final_connector_id = final_params.get("connector_id", connector.id)
final_delete_from_kb = final_params.get("delete_from_kb", delete_from_kb)
final_file_id = result.params.get("file_id", file_id)
final_connector_id = result.params.get("connector_id", connector.id)
final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
if final_connector_id != connector.id:
result = await db_session.execute(