Spaces:
Runtime error
Runtime error
File size: 3,353 Bytes
52f16ba c6421b9 4d860a2 57d78b2 4d860a2 57d78b2 33ffec6 52f16ba c6421b9 4d860a2 c6421b9 4d860a2 21f1bdf 4d860a2 c6421b9 4d860a2 c6421b9 33ffec6 c6421b9 4d860a2 c6421b9 4d860a2 57d78b2 4d860a2 52f16ba 57d78b2 4d860a2 c6421b9 4d860a2 c6421b9 4d860a2 c6421b9 57d78b2 33ffec6 57d78b2 4d860a2 c6421b9 57d78b2 c6421b9 33ffec6 57d78b2 c6421b9 4d860a2 57d78b2 4d860a2 52f16ba 4d860a2 c6421b9 4d860a2 52f16ba 4d860a2 b99e403 57d78b2 b99e403 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | from os import name
from langgraph.types import interrupt
from googleapiclient.errors import HttpError
from app.schemas.email_writing_agent_tools_schema import CreateDraftSchema, SendDraftSchema
from langchain.tools import tool
from langchain_google_community import GmailToolkit
from typing import Annotated, Union
from langchain_core.tools import InjectedToolCallId, tool
from langchain.tools import ToolRuntime
from langgraph.types import Command
from langgraph.graph import END
from langchain_core.messages import ToolMessage,SystemMessage, HumanMessage
from app.gmail_auth import gmail_toolkit ,api_resource
@tool(args_schema=CreateDraftSchema)
def create_gmail_draft(
to: Union[str, list],
subject: str,
body: str,
tool_call_id: Annotated[str, InjectedToolCallId] # Injected ID
):
"""Creates a new Gmail draft after human approval."""
if isinstance(to, list):
to = str(to[0]) if len(to) > 0 else "ERROR"
if to == "ERROR":
return "ERROR: 'to' parameter is empty."
# 1. Human-in-the-loop Interrupt
response = interrupt({
"action": "review_draft",
"data": {"to": to, "subject": subject, "body": body}
})
draft_tool = [t for t in gmail_toolkit.get_tools() if t.name == "create_gmail_draft"][0]
# 2. Handle Logic
if response.get("status") == "approved":
reply = draft_tool.invoke({"message": body, "to": [to], "subject": subject})
try:
draft_id = reply.split(":")[1].strip()
content = f"Draft created. User You MUST confirm before calling send_draft()."
# UPDATE STATE: Save draft_id directly
return Command(
update={
"draft_id": draft_id,
"reply_subject": subject,
"reply_email_body": body,
"messages": [ToolMessage(content, tool_call_id=tool_call_id)],
"name": "create_gmail_draft"
},goto=END
)
except IndexError:
return f"Draft created, but response parsing failed: {reply}"
else:
feedback = response.get("feedback", "User rejected.")
return f"DRAFT REJECTED BY USER. Feedback: {feedback}. Please rewrite."
#---------------------------------------------------------------------------
@tool
def send_draft(
tool_call_id: Annotated[str, InjectedToolCallId],runtime: ToolRuntime # Injected ID
):
"""Sends a finalized Gmail draft by its ID."""
try:
result = gmail_toolkit.api_resource.users().drafts().send(
userId="me", body={"id": runtime.state["draft_id"]}
).execute()
sent_id = result['id']
content = f"SUCCESS: Sent! a Gmail"
# UPDATE STATE: Save sent_message_id directly
return Command(
update={
"sent_message_id": sent_id,
"messages": [ToolMessage(content, tool_call_id=tool_call_id)],
"name": "send_draft"
}
)
except HttpError as error:
error_msg = f"ERROR: {error}"
return Command(
update={"messages": [ToolMessage(error_msg, tool_call_id=tool_call_id)], "name": "send_draft"}
)
email_writing_agent_tools = [
create_gmail_draft,
send_draft
] |