"""Shared jarvis-tools MCP endpoint (streamable-HTTP) for Railway. Mirrors the memory service's proven pattern so the SAME claude.ai custom connector setup works. Two doors, one handler, auth enforced BEFORE the MCP handler runs: • POST /mcp — Authorization: Bearer $JARVIS_TOOLS_KEY (Claude Code --header) • POST /mcp/ — token = first 32 hex of sha256(JARVIS_TOOLS_KEY) (path-secret door for claude.ai custom connectors, whose UI can't send custom headers; OAuth is overkill here). If JARVIS_TOOLS_KEY is empty, BOTH doors are open. Tools = the shared Google Drive folder (list/read/write/delete). Smartlead and the Ultron DB can be added here later as more @mcp.tool()s on the same connector. """ from __future__ import annotations import hashlib import hmac import os from contextlib import asynccontextmanager import anyio from mcp.server.fastmcp import FastMCP from mcp.server.transport_security import TransportSecuritySettings from starlette.routing import Route import asana import drive import emailbox import gworkspace import lookup import smartlead import storage import ultron_db API_KEY = os.environ.get("JARVIS_TOOLS_KEY", "") _BEARER = f"Bearer {API_KEY}" # claude.ai connectors can't send custom headers, so door #2 is a path secret: # first 32 hex of sha256(JARVIS_TOOLS_KEY). Print it with: # python3 -c "import hashlib,os;print(hashlib.sha256(os.environ['JARVIS_TOOLS_KEY'].encode()).hexdigest()[:32])" PATH_TOKEN = hashlib.sha256(API_KEY.encode()).hexdigest()[:32] if API_KEY else "" mcp = FastMCP( "jarvis-tools", stateless_http=True, json_response=True, # Railway's proxy makes FastMCP's localhost Host-validation 421 every # request; our doors already gate every call with a secret, so disable it. transport_security=TransportSecuritySettings( enable_dns_rebinding_protection=False), instructions=( "Ahmed's shared Jarvis storage — one Google Drive folder ('Jarvis " "Drive') any Claude on any device can read and write. drive_list to see " "what's there, drive_read to read a file (e.g. an Apollo CSV), " "drive_write to save one, drive_delete to remove one. It ONLY sees this " "one shared folder, never Ahmed's personal Drive. " "The asana_* tools are Ahmed's project management — everything he could do " "in the Asana web app: list/create/update/archive projects, invite people to " "them (asana_project_members) and post status updates; read a task in full " "(asana_task), list, create, update, move between projects/columns, " "complete, delete (recoverable 30 days) and break tasks into subtasks; set " "assignees, followers and tags; post and READ comment threads; and search. " "asana_api is the escape hatch for anything else in Asana's REST API. " "Prefer archiving a project over deleting it — archive is reversible."), ) @mcp.tool() async def drive_list() -> str: """List the files in the shared Jarvis Drive folder.""" return await anyio.to_thread.run_sync(drive.list_files) @mcp.tool() async def drive_read(file: str) -> str: """Read a text/csv/json file from the shared folder. `file` = name or id.""" return await anyio.to_thread.run_sync(lambda: drive.read_file(file)) @mcp.tool() async def drive_write(name: str, content: str) -> str: """Create or overwrite a text file in the shared folder from given content.""" return await anyio.to_thread.run_sync(lambda: drive.write_file(name, content)) @mcp.tool() async def drive_delete(file: str) -> str: """Remove (trash) a file from the shared folder. `file` = name or id.""" return await anyio.to_thread.run_sync(lambda: drive.delete_file(file)) # ---- Smartlead cold-email outreach ---------------------------------------- @mcp.tool() async def smartlead_campaigns() -> str: """List Ahmed's Smartlead cold-email campaigns and their status.""" return await anyio.to_thread.run_sync(smartlead.campaigns) @mcp.tool() async def smartlead_campaign_stats(campaign: str) -> str: """Analytics for one campaign (sent/open/reply/bounce). `campaign` = id or name.""" return await anyio.to_thread.run_sync(lambda: smartlead.campaign_stats(campaign)) @mcp.tool() async def smartlead_senders() -> str: """List sender inboxes and their warmup reputation / daily send limit.""" return await anyio.to_thread.run_sync(smartlead.senders) @mcp.tool() async def smartlead_create_campaign(name: str) -> str: """Create a new (drafted) campaign; returns its id.""" return await anyio.to_thread.run_sync(lambda: smartlead.create_campaign(name)) @mcp.tool() async def smartlead_set_sequence(campaign: str, sequence_json: str) -> str: """Set a campaign's email sequence. sequence_json = JSON array of steps {seq_number, seq_delay_details:{delay_in_days}, subject, email_body}.""" return await anyio.to_thread.run_sync( lambda: smartlead.set_sequence(campaign, sequence_json)) @mcp.tool() async def smartlead_add_leads(campaign: str, leads_json: str) -> str: """Add leads to a campaign. leads_json = JSON array of {email, first_name, last_name, company_name, ...}. Max 100 per call.""" return await anyio.to_thread.run_sync( lambda: smartlead.add_leads(campaign, leads_json)) @mcp.tool() async def smartlead_campaign_control(campaign: str, action: str) -> str: """Start / pause / stop a campaign. action = 'start' | 'pause' | 'stop'.""" return await anyio.to_thread.run_sync( lambda: smartlead.campaign_control(campaign, action)) @mcp.tool() async def smartlead_api(method: str, path: str, body_json: str = "") -> str: """Escape hatch — call ANY Smartlead REST endpoint. method=GET/POST/DELETE, path after /api/v1 (api_key added for you), body_json = JSON body for writes.""" return await anyio.to_thread.run_sync( lambda: smartlead.raw(method, path, body_json)) # ---- Ultron master DB (full Supabase SQL) --------------------------------- @mcp.tool() async def ultron_query(sql: str, confirm: str = "") -> str: """Run SQL against the Ultron master DB (18k+ leads + CRM: clients, contacts, users…). SELECT to read/aggregate, INSERT/UPDATE to write. Irreversible statements (drop/truncate, WHERE-less delete/update) need confirm='yes'.""" return await anyio.to_thread.run_sync(lambda: ultron_db.query(sql, confirm)) @mcp.tool() async def ultron_schema(table: str = "") -> str: """Inspect the DB: no table → all tables with row counts; with a table → its columns. Use before composing a query.""" return await anyio.to_thread.run_sync(lambda: ultron_db.schema(table)) # ---- Cross-source person / lead lookup ------------------------------------ @mcp.tool() async def person_lookup(query: str) -> str: """Search EVERYWHERE Ahmed keeps people/leads (memory graph, Ultron leads DB, Jarvis Drive sheets) by phone number or name. Use whenever Ahmed asks who someone is or gives an unknown number.""" return await anyio.to_thread.run_sync(lambda: lookup.person_lookup(query)) # ---- Business email — ahmad@revalstudio.com (Namecheap, SMTP+IMAP) --------- @mcp.tool() async def email_send(to: str, subject: str, body: str, cc: str = "") -> str: """Send an email from Ahmed's business mailbox (ahmad@revalstudio.com). `to` and `cc` are comma-separated; cc optional. Namecheap takes ~15-30s to deliver.""" return await anyio.to_thread.run_sync( lambda: emailbox.send(to, subject, body, cc)) @mcp.tool() async def email_search(query: str, limit: int = 10) -> str: """Search the business inbox by phrase (sender+subject; empty = latest). Returns a list with a UID per message for email_read.""" return await anyio.to_thread.run_sync(lambda: emailbox.search(query, limit)) @mcp.tool() async def email_read(uid: str) -> str: """Read the full body of one business-inbox message by its UID.""" return await anyio.to_thread.run_sync(lambda: emailbox.read(uid)) # ---- Work Google Workspace — ahmed.alrajeh@alrugaibfurniture.com ----------- @mcp.tool() async def gmail_send(to: str, subject: str, body: str, cc: str = "") -> str: """Send an email from Ahmed's WORK Gmail (alrugaib). to/cc comma-separated.""" return await anyio.to_thread.run_sync( lambda: gworkspace.gmail_send(to, subject, body, cc)) @mcp.tool() async def gmail_search(query: str, limit: int = 10) -> str: """Search the WORK Gmail inbox (Gmail query syntax; empty = latest). Returns a message id per result for gmail_read.""" return await anyio.to_thread.run_sync( lambda: gworkspace.gmail_search(query, limit)) @mcp.tool() async def gmail_read(message_id: str) -> str: """Read the full body of one WORK Gmail message by its id.""" return await anyio.to_thread.run_sync(lambda: gworkspace.gmail_read(message_id)) @mcp.tool() async def calendar_list(days: int = 7) -> str: """List Ahmed's WORK calendar events for the next N days (default 7).""" return await anyio.to_thread.run_sync(lambda: gworkspace.calendar_list(days)) @mcp.tool() async def calendar_create_event(title: str, start: str, end: str, attendees: str = "", description: str = "", add_meet: str = "") -> str: """Create a WORK calendar event. start/end = ISO datetime with timezone (2026-07-15T14:00:00+03:00). attendees = comma-separated emails (invited). add_meet='yes' attaches a Google Meet link.""" return await anyio.to_thread.run_sync( lambda: gworkspace.calendar_create_event( title, start, end, attendees, description, add_meet)) @mcp.tool() async def sheets_read(spreadsheet_id: str, a1_range: str) -> str: """Read a range from a Google Sheet (work account). a1_range e.g. 'Sheet1!A1:D20'.""" return await anyio.to_thread.run_sync( lambda: gworkspace.sheets_read(spreadsheet_id, a1_range)) @mcp.tool() async def sheets_write(spreadsheet_id: str, a1_range: str, values_json: str) -> str: """Write to a Google Sheet range (work account). values_json = JSON array of rows, e.g. [["a","b"],["c","d"]].""" return await anyio.to_thread.run_sync( lambda: gworkspace.sheets_write(spreadsheet_id, a1_range, values_json)) # ---- Shared WRITABLE storage (Supabase Storage) --------------------------- @mcp.tool() async def storage_write(name: str, content: str) -> str: """Save a text file to the shared writable storage (syncs across all devices). `name` = file name/path (e.g. 'crm/riyadh.csv'), `content` = the full text. This is where Jarvis/AIs put files for each other (drive_* is read-only).""" return await anyio.to_thread.run_sync(lambda: storage.write(name, content)) @mcp.tool() async def storage_read(name: str) -> str: """Read a file back from the shared writable storage by name/path.""" return await anyio.to_thread.run_sync(lambda: storage.read(name)) @mcp.tool() async def storage_list() -> str: """List the files in the shared writable storage.""" return await anyio.to_thread.run_sync(storage.list_files) @mcp.tool() async def storage_delete(name: str) -> str: """Delete a file from the shared writable storage by name/path.""" return await anyio.to_thread.run_sync(lambda: storage.delete(name)) # ---- Asana project management --------------------------------------------- # One implementation, two doors: these wrap the SAME asana.py that voice Jarvis # imports (voice/asana_control.py), so both have identical capability. @mcp.tool() async def asana_workspaces() -> str: """List Ahmed's Asana workspaces and which one the tools default to.""" return await anyio.to_thread.run_sync(asana.workspaces) @mcp.tool() async def asana_projects(query: str = "", archived: bool = False) -> str: """List Asana projects; `query` filters by name, archived=True lists archived ones. Use it to find the right project before asking for its tasks.""" return await anyio.to_thread.run_sync(lambda: asana.projects(query, archived)) @mcp.tool() async def asana_project_create(name: str, team: str = "", notes: str = "", due_on: str = "", privacy: str = "") -> str: """Create an Asana project. `team` is required only if the workspace is an ORGANIZATION (the tool says so, and names the teams, if it is). `privacy` = public_to_workspace | private_to_team | private.""" return await anyio.to_thread.run_sync( lambda: asana.project_create(name, team, notes, due_on, privacy)) @mcp.tool() async def asana_project_update(project: str, name: str = "", notes: str = "", due_on: str = "", owner: str = "", archive: bool | None = None) -> str: """Change a project (`project` = name or id): rename, notes, due_on, owner, or archive=True to ARCHIVE it (reversible, keeps everything — this is how you retire a project; deleting one is not offered here because it can't be undone, use asana_api if you truly mean DELETE).""" return await anyio.to_thread.run_sync( lambda: asana.project_update(project, name, notes, due_on, owner, archive)) @mcp.tool() async def asana_project_members(project: str, add: str = "", remove: str = "", access_level: str = "") -> str: """Who's on a project, and invite/remove them. `add`/`remove` = comma-separated emails, names or ids ('me' works). `access_level` = admin|editor|commenter|viewer. No add/remove = list them. This shares a project with someone already in the workspace; inviting a NEW person to Asana itself is asana_api POST /workspaces/{gid}/addUser.""" return await anyio.to_thread.run_sync( lambda: asana.project_members(project, add, remove, access_level)) @mcp.tool() async def asana_project_status(project: str, text: str, status_type: str = "on_track", title: str = "") -> str: """Post a project status update. status_type = on_track | at_risk | off_track | on_hold | complete.""" return await anyio.to_thread.run_sync( lambda: asana.project_status(project, text, status_type, title)) @mcp.tool() async def asana_task(task: str) -> str: """FULL detail of ONE task (`task` = name or id): notes, assignee, due date, project, section, tags, followers, subtasks and the latest comments.""" return await anyio.to_thread.run_sync(lambda: asana.task(task)) @mcp.tool() async def asana_tasks(project: str = "", mine: bool = True, section: str = "", completed: bool = False, limit: int = 50) -> str: """List Asana tasks. `project` (name or id) = that project's; add `section` for one board column; empty project with mine=True = 'what's on my plate'. Open tasks only unless completed=True.""" return await anyio.to_thread.run_sync( lambda: asana.tasks(project, mine, section, completed, limit)) @mcp.tool() async def asana_task_create(name: str, notes: str = "", project: str = "", assignee: str = "me", due_on: str = "", section: str = "", parent: str = "") -> str: """Create an Asana task. `project` = project name or id (optional), `section` = a board column in it, `parent` = another task to make this a SUBTASK of, `due_on` = YYYY-MM-DD, `assignee` = 'me' or a user id/email/name.""" return await anyio.to_thread.run_sync( lambda: asana.task_create(name, notes, project, assignee, due_on, section, parent)) @mcp.tool() async def asana_task_update(task: str, name: str = "", notes: str = "", due_on: str = "", start_on: str = "", assignee: str = "", complete: bool | None = None) -> str: """Change a task (`task` = name or id): rename, edit notes, set due_on (YYYY-MM-DD), reassign, or set complete true/false. `start_on` needs a due date too AND a paid Asana plan (it 402s on a free workspace).""" return await anyio.to_thread.run_sync( lambda: asana.task_update(task, name, notes, due_on, start_on, assignee, complete)) @mcp.tool() async def asana_task_move(task: str, project: str = "", section: str = "", remove_from: str = "") -> str: """Move a task to another `project` and/or `section`. A REAL move: it is removed from the project it was in (Asana's addProject alone would leave it in both). `section` alone moves it between columns of its current project.""" return await anyio.to_thread.run_sync( lambda: asana.task_move(task, project, section, remove_from)) @mcp.tool() async def asana_task_complete(task: str) -> str: """Mark an Asana task done. `task` = its name or id.""" return await anyio.to_thread.run_sync(lambda: asana.task_complete(task)) @mcp.tool() async def asana_task_delete(task: str) -> str: """Delete an Asana task. It goes to Deleted Items and is recoverable for 30 days. Only when he means DELETE — 'it's done' is asana_task_complete.""" return await anyio.to_thread.run_sync(lambda: asana.task_delete(task)) @mcp.tool() async def asana_task_people(task: str, assignee: str = "", add_followers: str = "", remove_followers: str = "", add_tag: str = "", remove_tag: str = "") -> str: """Who/what is on a task: set the assignee, add/remove followers (comma-separated names, emails or 'me'), add/remove tags (a new tag is created if needed). With no changes it just reports.""" return await anyio.to_thread.run_sync( lambda: asana.task_people(task, assignee, add_followers, remove_followers, add_tag, remove_tag)) @mcp.tool() async def asana_subtasks(task: str, add: str = "") -> str: """List a task's subtasks, or create them: `add` = comma-separated subtask names.""" return await anyio.to_thread.run_sync(lambda: asana.subtasks(task, add)) @mcp.tool() async def asana_comment(task: str, text: str) -> str: """Post a comment on an Asana task (`task` = name or id) — e.g. a status update.""" return await anyio.to_thread.run_sync(lambda: asana.comment(task, text)) @mcp.tool() async def asana_comments(task: str, limit: int = 10) -> str: """READ a task's comment thread (system events skipped). `limit` = how many of the latest (default 10).""" return await anyio.to_thread.run_sync(lambda: asana.comments(task, limit)) @mcp.tool() async def asana_search(query: str = "", project: str = "", assignee: str = "", due_before: str = "", due_after: str = "", completed: bool | None = None) -> str: """Find tasks by phrase, with optional filters (project, assignee='me' or a name, due_before/due_after as YYYY-MM-DD, completed). Uses Asana's premium search when the plan has it, else name-matching typeahead. The search index lags up to a minute — for a task just created, use asana_task / asana_tasks instead.""" return await anyio.to_thread.run_sync( lambda: asana.search(query, project, assignee, due_before, due_after, completed)) @mcp.tool() async def asana_api(method: str, path: str, body_json: str = "", params_json: str = "") -> str: """Escape hatch — call ANY Asana REST endpoint (base https://app.asana.com/api/1.0) for what the curated asana_* tools don't cover: sections, tags, teams, attachments, dependencies, templates, goals, portfolios, custom fields, webhooks, batch, inviting a NEW person to the workspace, deleting a project. method = GET/POST/PUT/DELETE. path = everything after /api/1.0, e.g. '/projects/12345/sections'. ENVELOPE: do NOT wrap body_json in {"data": ...} — that's added for you. Just pass the inner object: {"name": "Q3 launch", "workspace": "12345"}. The response is unwrapped from `data` for you too. OPT_FIELDS (important): Asana returns ONLY gid/resource_type/name by default. To get real fields, pass them in params_json: params_json = {"opt_fields": "name,due_on,completed,assignee.name", "limit": 50} Dot-notation reaches into related objects (assignee.name, projects.name). Returns raw JSON.""" return await anyio.to_thread.run_sync( lambda: asana.raw(method, path, body_json, params_json)) _ = mcp.streamable_http_app() session_manager = mcp.session_manager @asynccontextmanager async def lifespan(): """Enter from the FastAPI lifespan so the session manager's task group runs.""" async with session_manager.run(): yield def _authorized(scope, token: str | None) -> bool: if not API_KEY: return True if token is not None: return hmac.compare_digest(token, PATH_TOKEN) headers = {k.decode().lower(): v.decode() for k, v in scope.get("headers", [])} return hmac.compare_digest(headers.get("authorization", ""), _BEARER) async def _reject(send) -> None: await send({"type": "http.response.start", "status": 401, "headers": [(b"content-type", b"application/json")]}) await send({"type": "http.response.body", "body": b'{"error":"unauthorized"}'}) class _Door: """ASGI gate in front of the shared MCP handler. mode: 'header' or 'token'.""" def __init__(self, mode: str): self.mode = mode async def __call__(self, scope, receive, send): if scope["type"] != "http": await _reject(send) return token = (scope.get("path_params", {}).get("token") if self.mode == "token" else None) if not _authorized(scope, token): await _reject(send) return await session_manager.handle_request(scope, receive, send) def mount(app) -> None: """Attach both MCP doors to an existing FastAPI / Starlette app.""" app.router.routes.append(Route("/mcp", _Door("header"))) app.router.routes.append(Route("/mcp/{token}", _Door("token")))