Spaces:
Running
Running
| """ | |
| Gmail Agent – specialized agent with only Gmail MCP tools. | |
| Uses MCPServerStdio (local subprocess) with create_static_tool_filter so the | |
| agent sees *only* email tools. Zero network overhead. | |
| """ | |
| import asyncio | |
| import logging | |
| from openai import AsyncOpenAI | |
| from agents import Agent, Runner, OpenAIChatCompletionsModel | |
| from agents.model_settings import ModelSettings | |
| try: | |
| from .google_mcp_config import ( | |
| LONGCAT_API_KEY, LONGCAT_BASE_URL, MODEL_NAME, | |
| GMAIL_TOOLS, create_google_mcp_server, USER_GOOGLE_EMAIL, | |
| ) | |
| except ImportError: | |
| from google_mcp_config import ( | |
| LONGCAT_API_KEY, LONGCAT_BASE_URL, MODEL_NAME, | |
| GMAIL_TOOLS, create_google_mcp_server, USER_GOOGLE_EMAIL, | |
| ) | |
| logger = logging.getLogger(__name__) | |
| SYSTEM_PROMPT = """\ | |
| You are a specialized Gmail assistant. You can search, read, compose, | |
| label, filter, and manage emails using the available tools. | |
| Capabilities: | |
| - **Search** messages (Gmail search syntax, e.g. "from:alice subject:invoice") | |
| - **Read** individual messages or full threads (including batch reads) | |
| - **Download** file attachments from messages | |
| - **Send** emails (with To, CC, BCC, subject, body) | |
| - **Draft** emails for later review | |
| - **Labels** — list, create, update, or delete labels; apply or remove | |
| labels on messages (single or batch) | |
| - **Filters** — list, create, or delete inbox filters (auto-label, | |
| auto-archive, auto-forward, etc.) | |
| - **Threads** — read full conversation threads (single or batch) | |
| Rules: | |
| 1. The user's Google email is provided in the query — use it for every | |
| tool call in the `user_google_email` parameter. NEVER ask the user | |
| for their email; it is always supplied. | |
| 2. Before sending an email, show a preview of the composed message | |
| (recipients, subject, body) and ask the user to confirm unless they | |
| explicitly said "just send it". | |
| 3. When reading messages, include sender, subject, date, and a brief | |
| excerpt. For threads, note the number of messages. | |
| 4. For label- or filter-management, confirm the action before and after. | |
| 5. Use standard Gmail search operators where possible: | |
| `from:`, `to:`, `subject:`, `has:attachment`, `after:`, `before:`, | |
| `is:unread`, `label:`, `filename:`, etc. | |
| """ | |
| class GmailAgent: | |
| """Thin wrapper around the OpenAI Agent SDK wired to Gmail tools.""" | |
| def __init__(self, model: str = MODEL_NAME): | |
| self.model = model | |
| self._client = AsyncOpenAI( | |
| api_key=LONGCAT_API_KEY, | |
| base_url=LONGCAT_BASE_URL, | |
| timeout=30.0, | |
| ) | |
| logger.info("[GMAIL-AGENT] __init__ complete") | |
| logger.info("[GMAIL-AGENT] model : %s", self.model) | |
| logger.info("[GMAIL-AGENT] base_url : %s", LONGCAT_BASE_URL) | |
| logger.info("[GMAIL-AGENT] api_key : %s...", (LONGCAT_API_KEY or "")[:12]) | |
| # ── factory helpers ────────────────────────────────────────────────── | |
| def _create_mcp_server(self): | |
| """Spawn a local MCP subprocess with only Gmail tools loaded.""" | |
| logger.info("[GMAIL-AGENT] Creating MCP server (stdio subprocess)...") | |
| server = create_google_mcp_server(service="gmail", tool_names=GMAIL_TOOLS) | |
| logger.info("[GMAIL-AGENT] MCP server object created: %s", server) | |
| return server | |
| def _create_agent(self, mcp_server) -> Agent: | |
| logger.info("[GMAIL-AGENT] Creating inner Agent with mcp_servers=[%s]", mcp_server) | |
| agent = Agent( | |
| name="Gmail Agent", | |
| instructions=SYSTEM_PROMPT, | |
| mcp_servers=[mcp_server], | |
| model=OpenAIChatCompletionsModel( | |
| model=self.model, | |
| openai_client=self._client, | |
| ), | |
| model_settings=ModelSettings(tool_choice="auto"), | |
| ) | |
| logger.info("[GMAIL-AGENT] Inner Agent created: name=%s, tools=%s, mcp_servers=%d", | |
| agent.name, agent.tools, len(agent.mcp_servers)) | |
| return agent | |
| # ── public API ─────────────────────────────────────────────────────── | |
| async def run(self, query: str) -> str: | |
| """Spawn MCP connection, run a single query, then clean up.""" | |
| import time as _time | |
| logger.info("="*60) | |
| logger.info("[GMAIL-AGENT] run() called with query: %s", query[:200]) | |
| logger.info("="*60) | |
| # Step 1: Create MCP server object | |
| logger.info("[GMAIL-AGENT] STEP 1: Creating MCP server object...") | |
| t0 = _time.time() | |
| mcp_server = self._create_mcp_server() | |
| logger.info("[GMAIL-AGENT] STEP 1 DONE (%.2fs)", _time.time() - t0) | |
| # Step 2: Connect to MCP server (start subprocess) | |
| logger.info("[GMAIL-AGENT] STEP 2: Entering 'async with mcp_server:' (starting subprocess)...") | |
| t1 = _time.time() | |
| async with mcp_server: | |
| logger.info("[GMAIL-AGENT] STEP 2 DONE: MCP subprocess started (%.2fs)", _time.time() - t1) | |
| # Step 3: List tools from MCP server | |
| logger.info("[GMAIL-AGENT] STEP 3: Listing tools from MCP server...") | |
| t2 = _time.time() | |
| try: | |
| tools = await mcp_server.list_tools() | |
| logger.info("[GMAIL-AGENT] STEP 3 DONE: Got %d tools (%.2fs)", len(tools), _time.time() - t2) | |
| for i, tool in enumerate(tools): | |
| logger.info("[GMAIL-AGENT] Tool[%d]: %s", i, getattr(tool, 'name', tool)) | |
| except Exception as e: | |
| logger.error("[GMAIL-AGENT] STEP 3 FAILED: list_tools error: %s", e, exc_info=True) | |
| # Step 4: Create inner Agent | |
| logger.info("[GMAIL-AGENT] STEP 4: Creating inner Agent...") | |
| agent = self._create_agent(mcp_server) | |
| logger.info("[GMAIL-AGENT] STEP 4 DONE: Agent ready") | |
| # Step 5: Call Runner.run() | |
| logger.info("[GMAIL-AGENT] STEP 5: Calling Runner.run(agent, query)...") | |
| logger.info("[GMAIL-AGENT] Agent name : %s", agent.name) | |
| logger.info("[GMAIL-AGENT] Agent model : %s", self.model) | |
| logger.info("[GMAIL-AGENT] Agent tool_choice: auto") | |
| logger.info("[GMAIL-AGENT] Query : %s", query[:200]) | |
| t3 = _time.time() | |
| try: | |
| result = await Runner.run(agent, input=query) | |
| logger.info("[GMAIL-AGENT] STEP 5 DONE: Runner.run() completed (%.2fs)", _time.time() - t3) | |
| logger.info("[GMAIL-AGENT] final_output length: %d", len(result.final_output or "")) | |
| logger.info("[GMAIL-AGENT] final_output preview: %s", (result.final_output or "")[:300]) | |
| # Log new_items to see what the model actually did | |
| if hasattr(result, 'new_items'): | |
| logger.info("[GMAIL-AGENT] new_items count: %d", len(result.new_items)) | |
| for idx, item in enumerate(result.new_items): | |
| logger.info("[GMAIL-AGENT] new_items[%d]: type=%s", idx, type(item).__name__) | |
| return result.final_output | |
| except Exception as e: | |
| logger.error("[GMAIL-AGENT] STEP 5 FAILED: Runner.run() error: %s", e, exc_info=True) | |
| raise | |
| # ─── CLI entry point ────────────────────────────────────────────────────────── | |
| async def main(): | |
| agent = GmailAgent() | |
| resp = await agent.run( | |
| "Search for unread emails from last week. My email is user@example.com" | |
| ) | |
| print("Agent Response:\n", resp) | |
| if __name__ == "__main__": | |
| from dotenv import load_dotenv, find_dotenv | |
| load_dotenv(find_dotenv()) | |
| asyncio.run(main()) | |