File size: 4,063 Bytes
3370983 1b2d294 3370983 1b2d294 3370983 1b2d294 3370983 1b2d294 3370983 |
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 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
import asyncio
import shutil
from pathlib import Path
from langchain_core.tools import tool
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from src.mcp_servers.examples.gmail.settings import GMailSettings
from src.backend.prompts import get_prompt
# Attempt to find uv executable
#-----------------------------------------------------------------------------
# `Dockerfile.supervisor` installs uv in the base image in `/usr/local/bin/uv`
# `which` attempts to find it in the system PATH and returns the full path to it.
UV_PATH = shutil.which("uv")
SYSTEM_PROMPT = get_prompt(
template_name="GMail",
latest_version=True
)
@tool
def gmail_agent(query: str) -> str:
"""
A tool that acts as a Gmail agent.
It can read, search, label, and send emails using the Gmail MCP server.
Args:
query (str): The natural language request (e.g., "Send an email to X", "Check unread emails").
Returns:
str: The natural language response from the agent confirming the action or providing the requested information.
Example output:
"I have successfully sent the email to X with the subject 'Interview Invitation'."
"""
if not UV_PATH:
return "❌ Error: 'uv' executable not found. Please ensure uv is installed and in the system PATH."
# Validate required Gmail settings before proceeding
import os
creds_env = os.getenv("GMAIL_CREDS_JSON")
token_env = os.getenv("GMAIL_TOKEN_JSON")
if not creds_env or not token_env:
return "❌ Gmail credentials not configured. Set GMAIL_CREDS_JSON and GMAIL_TOKEN_JSON environment variables."
try:
import asyncio
from pydantic_core import ValidationError
async def _run_async():
# Load settings
try:
# Pass env values directly to avoid reliance on env file loading
settings = GMailSettings(creds_json=creds_env, token_json=token_env)
except ValidationError as ve:
return f"❌ Gmail settings invalid: {ve}"
except Exception as e:
return f"❌ Gmail settings error: {e}"
creds_path, token_path = settings.materialize_files()
# Initialize model
model = ChatOpenAI(model="gpt-4o", temperature=0)
# Connect to MCP server
client = MultiServerMCPClient(
{
"gmail": {
"command": UV_PATH,
"args": [
"--directory", str(settings.gmail_mcp_dir),
"run", "gmail",
"--creds-file-path", str(creds_path),
"--token-path", str(token_path),
],
"transport": "stdio",
}
}
)
# Fetch tools
try:
tools = await client.get_tools()
except Exception as e:
return f"❌ Failed to connect to Gmail MCP server: {str(e)}"
if not tools:
return "❌ No tools available from Gmail MCP server."
# Create agent
agent = create_agent(model, tools)
# Run agent
result = await agent.ainvoke({
"messages": [
{
"role": "system",
"content": SYSTEM_PROMPT,
},
{
"role": "user",
"content": query,
},
]
})
# Extract result
output = result["messages"][-1].content
return output
return asyncio.run(_run_async())
except Exception as e:
import traceback
return f"❌ Error in gmail_agent: {str(e)}\n{traceback.format_exc()}"
|