mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-09 11:17:44 +02:00
[eric] split: extract tools_lib mcp-config/discovery/oauth helpers
This commit is contained in:
ciregenz
@@ -0,0 +1,294 @@
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import sys
|
||||
from typing import Optional
|
||||
|
||||
from backend.apps.tools_lib.models import ToolDefinition
|
||||
from backend.apps.tools_lib.oauth_config import OPENSWARM_OAUTH_BASE_URL
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _sanitize_server_name(name: str) -> str:
|
||||
"""Convert a tool name into a valid MCP server identifier (alphanumeric + hyphens)."""
|
||||
return re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
|
||||
|
||||
|
||||
def _extra_bin_dirs() -> list[str]:
|
||||
"""Well-known user-local bin directories that may not be on PATH in packaged apps."""
|
||||
home = os.path.expanduser("~")
|
||||
# Bundled uv-bin (ships uvx for non-dev users)
|
||||
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
dirs = [
|
||||
os.path.join(_backend, "uv-bin"),
|
||||
os.path.join(home, ".bun", "bin"),
|
||||
os.path.join(home, ".cargo", "bin"),
|
||||
os.path.join(home, ".local", "bin"),
|
||||
os.path.join(home, ".volta", "bin"),
|
||||
"/opt/homebrew/bin",
|
||||
"/usr/local/bin",
|
||||
]
|
||||
# nvm: pick the newest installed node version
|
||||
nvm_node = os.path.join(home, ".nvm", "versions", "node")
|
||||
try:
|
||||
if os.path.isdir(nvm_node):
|
||||
versions = sorted(os.listdir(nvm_node), reverse=True)
|
||||
if versions:
|
||||
dirs.insert(0, os.path.join(nvm_node, versions[0], "bin"))
|
||||
except OSError:
|
||||
pass
|
||||
# fnm
|
||||
fnm_bin = os.path.join(home, "Library", "Application Support", "fnm", "aliases", "default", "bin")
|
||||
if os.path.isdir(fnm_bin):
|
||||
dirs.insert(0, fnm_bin)
|
||||
return dirs
|
||||
|
||||
|
||||
def _resolve_command(command: str) -> str | None:
|
||||
"""Find a command on PATH, falling back to common user-local bin directories
|
||||
and bundled binaries (uv-bin for uvx/uv)."""
|
||||
found = shutil.which(command)
|
||||
if found:
|
||||
return found
|
||||
# Windows binaries need an extension. shutil.which() handles PATHEXT for
|
||||
# PATH lookups, but we manually scan _extra_bin_dirs below; replicate
|
||||
# the suffix probing here so `uvx` finds `uvx.exe`, etc.
|
||||
if sys.platform == "win32":
|
||||
suffixes = [""] + os.environ.get("PATHEXT", ".COM;.EXE;.BAT;.CMD").lower().split(os.pathsep)
|
||||
else:
|
||||
suffixes = [""]
|
||||
def _probe(directory: str) -> str | None:
|
||||
for suffix in suffixes:
|
||||
candidate = os.path.join(directory, command + suffix)
|
||||
if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
|
||||
return candidate
|
||||
return None
|
||||
for d in _extra_bin_dirs():
|
||||
hit = _probe(d)
|
||||
if hit:
|
||||
return hit
|
||||
# Check bundled uv-bin directory (ships uv/uvx for non-dev users)
|
||||
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
return _probe(os.path.join(_backend, "uv-bin"))
|
||||
|
||||
|
||||
def _augmented_path() -> str:
|
||||
"""Return PATH with extra bin dirs prepended (for child process environments)."""
|
||||
extra = [d for d in _extra_bin_dirs() if os.path.isdir(d)]
|
||||
current = os.environ.get("PATH", "")
|
||||
seen: set[str] = set()
|
||||
parts: list[str] = []
|
||||
for p in extra + current.split(os.pathsep):
|
||||
if p and p not in seen:
|
||||
seen.add(p)
|
||||
parts.append(p)
|
||||
return os.pathsep.join(parts)
|
||||
|
||||
|
||||
def derive_mcp_config(tool: ToolDefinition) -> Optional[dict]:
|
||||
"""Build the claude_agent_sdk mcp_servers config entry for a tool.
|
||||
|
||||
Returns None if the tool cannot be configured (e.g. missing data).
|
||||
"""
|
||||
if not tool.mcp_config:
|
||||
return None
|
||||
|
||||
config: dict = dict(tool.mcp_config)
|
||||
|
||||
if tool.credentials:
|
||||
if config.get("type") in ("http", "sse"):
|
||||
headers = config.setdefault("headers", {})
|
||||
for key, val in tool.credentials.items():
|
||||
if key.lower() in ("authorization", "api_key", "api-key"):
|
||||
headers.setdefault("Authorization", f"Bearer {val}")
|
||||
else:
|
||||
env = config.setdefault("env", {})
|
||||
env.update(tool.credentials)
|
||||
|
||||
if tool.oauth_tokens.get("access_token"):
|
||||
if config.get("type") in ("http", "sse"):
|
||||
headers = config.setdefault("headers", {})
|
||||
headers["Authorization"] = f"Bearer {tool.oauth_tokens['access_token']}"
|
||||
else:
|
||||
env = config.setdefault("env", {})
|
||||
env["OAUTH_ACCESS_TOKEN"] = tool.oauth_tokens["access_token"]
|
||||
if tool.name.lower() == "notion":
|
||||
env["NOTION_TOKEN"] = tool.oauth_tokens["access_token"]
|
||||
if tool.name.lower() == "hubspot":
|
||||
env["PRIVATE_APP_ACCESS_TOKEN"] = tool.oauth_tokens["access_token"]
|
||||
if tool.oauth_tokens.get("refresh_token"):
|
||||
env["GOOGLE_WORKSPACE_REFRESH_TOKEN"] = tool.oauth_tokens["refresh_token"]
|
||||
# google_workspace_mcp's gauth.py hardcodes token_uri to
|
||||
# https://oauth2.googleapis.com/token and refreshes using the
|
||||
# local CLIENT_ID/SECRET on every API call. The OAuth flow
|
||||
# itself runs through the cloud's rotation pool, so the
|
||||
# refresh_token is bound to whichever pool slot minted it,
|
||||
# not the single client baked into the DMG. Mismatch -> Google
|
||||
# returns unauthorized_client. We point token_uri at a local
|
||||
# proxy that forwards the refresh to our cloud's pool-aware
|
||||
# /api/oauth/google/refresh endpoint; CLIENT_ID/SECRET become
|
||||
# unused placeholders (gauth.py only validates non-empty).
|
||||
_port = os.environ.get("OPENSWARM_PORT", "8324")
|
||||
env["GOOGLE_WORKSPACE_TOKEN_URI"] = (
|
||||
f"http://127.0.0.1:{_port}/api/tools/google-oauth-token"
|
||||
)
|
||||
env.setdefault("GOOGLE_WORKSPACE_CLIENT_ID", "openswarm-proxy")
|
||||
env.setdefault("GOOGLE_WORKSPACE_CLIENT_SECRET", "openswarm-proxy")
|
||||
|
||||
# Google Workspace MCP: redirect spawn through our shim that
|
||||
# monkey-patches gauth.get_credentials before the worker registers
|
||||
# tools, so token_uri points at our local proxy. Stays a stdio
|
||||
# subprocess; google-workspace-mcp gets installed into uv's
|
||||
# ephemeral env via --with, same way the upstream entry-point
|
||||
# invocation used to do it.
|
||||
if tool.name.lower() == "google workspace" and config.get("type") == "stdio":
|
||||
shim_path = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
"google_workspace_mcp_shim",
|
||||
"run.py",
|
||||
)
|
||||
config["command"] = "uv"
|
||||
config["args"] = ["run", "--with", "google-workspace-mcp", "python", shim_path]
|
||||
|
||||
# Discord MCP runs as a small Python shim (backend.apps.discord_mcp_shim).
|
||||
# We pass install_id + base URL via env so the shim subprocess doesn't
|
||||
# need to import backend.config.* itself.
|
||||
if tool.name.lower() == "discord" and config.get("type") == "stdio":
|
||||
from backend.config.install_id import get_install_id
|
||||
env = config.setdefault("env", {})
|
||||
env["OPENSWARM_OAUTH_BASE_URL"] = OPENSWARM_OAUTH_BASE_URL
|
||||
env["OPENSWARM_INSTALL_ID"] = get_install_id()
|
||||
# Pass the authorized guild IDs so the shim can scope-enforce.
|
||||
guild_ids = [g.get("id", "") for g in (tool.oauth_tokens.get("guilds") or []) if g.get("id")]
|
||||
if guild_ids:
|
||||
env["OPENSWARM_DISCORD_GUILD_IDS"] = ",".join(guild_ids)
|
||||
# The shim runs as a subprocess and needs to import
|
||||
# `backend.apps.discord_mcp_shim`; set PYTHONPATH to the project
|
||||
# root (parent of the backend/ dir) so that import resolves.
|
||||
_project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
||||
existing_pp = env.get("PYTHONPATH") or os.environ.get("PYTHONPATH", "")
|
||||
env["PYTHONPATH"] = (_project_root + os.pathsep + existing_pp) if existing_pp else _project_root
|
||||
|
||||
# Microsoft 365 MCP: use a stable token cache path shared across process spawns
|
||||
if tool.name.lower() == "microsoft 365" and config.get("type") == "stdio":
|
||||
env = config.setdefault("env", {})
|
||||
cache_dir = os.path.join(os.path.expanduser("~"), ".openswarm")
|
||||
os.makedirs(cache_dir, exist_ok=True)
|
||||
env["MS365_MCP_TOKEN_CACHE_PATH"] = os.path.join(cache_dir, "ms365-token-cache.json")
|
||||
env["MS365_MCP_SELECTED_ACCOUNT_PATH"] = os.path.join(cache_dir, "ms365-selected-account.json")
|
||||
|
||||
if config.get("type") == "stdio":
|
||||
if config.get("command"):
|
||||
# `python` (no version suffix) doesn't exist on a stock macOS,
|
||||
# so a tool config that asks for "python" silently fails to
|
||||
# spawn; Claude Agent SDK then exposes zero tools from that
|
||||
# MCP. We resolve to the actual interpreter running the
|
||||
# backend (sys.executable), which is guaranteed to exist and
|
||||
# have backend modules importable. `python3` and absolute
|
||||
# paths pass through unchanged.
|
||||
if config["command"] == "python":
|
||||
resolved_python = sys.executable or shutil.which("python3") or shutil.which("python")
|
||||
if resolved_python:
|
||||
config["command"] = resolved_python
|
||||
# Check for bundled npm MCP servers; use Electron's Node.js instead of npx
|
||||
if config["command"] in ("npx", "bunx"):
|
||||
pkg_name = next((a for a in (config.get("args") or []) if not a.startswith("-")), None)
|
||||
if pkg_name:
|
||||
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
electron_path = os.environ.get("OPENSWARM_ELECTRON_PATH")
|
||||
# Two bundle layouts in mcp-bundles/, checked in priority order:
|
||||
#
|
||||
# 1. Multi-file bundle dir: mcp-bundles/<safe>/dist/index.js
|
||||
# Used when the SDK reads sibling files at runtime.
|
||||
# Examples: @softeria/ms-365-mcp-server reads
|
||||
# ../package.json for --version and dist/endpoints.json
|
||||
# for Graph API definitions; @notionhq/notion-mcp-server
|
||||
# reads ../scripts/notion-openapi.json. The build script
|
||||
# ships a stripped package.json (no "type":"module") next
|
||||
# to dist/ so __dirname/../package.json resolves correctly.
|
||||
# See scripts/build-app.sh `build_mcp_bundle_dir`.
|
||||
#
|
||||
# 2. Single-file bundle: mcp-bundles/<safe>.js
|
||||
# Used when the SDK is fully self-contained
|
||||
# (reddit-mcp-buddy).
|
||||
#
|
||||
# Scoped names get flattened ("@softeria/ms-365-mcp-server"
|
||||
# -> "softeria-ms-365-mcp-server") for filesystem safety.
|
||||
safe_bundle = pkg_name.replace("/", "-").replace("@", "")
|
||||
bundle_dir_path = os.path.join(_backend, "mcp-bundles", safe_bundle, "dist", "index.js")
|
||||
bundle_file_path = os.path.join(_backend, "mcp-bundles", f"{safe_bundle}.js")
|
||||
bundle_path = None
|
||||
if os.path.isfile(bundle_dir_path):
|
||||
bundle_path = bundle_dir_path
|
||||
elif os.path.isfile(bundle_file_path):
|
||||
bundle_path = bundle_file_path
|
||||
# Prefer the bundled real-Node binary over Electron-as-Node:
|
||||
# avoids the bouncing "exec" Dock icon on fresh user Macs +
|
||||
# spawns ~10x faster than re-execing the OpenSwarm Electron
|
||||
# binary as Node. Falls back to Electron-as-Node only if
|
||||
# the bundled node payload wasn't shipped (legacy builds).
|
||||
bundled_node = os.environ.get("OPENSWARM_NODE_PATH")
|
||||
if bundle_path and bundled_node and os.path.exists(bundled_node):
|
||||
config["command"] = bundled_node
|
||||
config["args"] = [bundle_path]
|
||||
logger.info(f"Using bundled MCP server for {pkg_name} via bundled node ({bundle_path})")
|
||||
elif bundle_path and electron_path:
|
||||
config["command"] = electron_path
|
||||
config["args"] = [bundle_path]
|
||||
config.setdefault("env", {})["ELECTRON_RUN_AS_NODE"] = "1"
|
||||
logger.info(f"Using bundled MCP server for {pkg_name} ({bundle_path})")
|
||||
else:
|
||||
# Check for pre-installed npm package (works in both dev and packaged modes)
|
||||
safe_dir = pkg_name.replace("/", "-").replace("@", "")
|
||||
npm_dir = os.path.join(_backend, "npm-servers", safe_dir)
|
||||
pkg_json_path = os.path.join(npm_dir, "node_modules", pkg_name, "package.json")
|
||||
if os.path.isfile(pkg_json_path):
|
||||
import json as _json
|
||||
with open(pkg_json_path) as f:
|
||||
pkg_meta = _json.load(f)
|
||||
bin_field = pkg_meta.get("bin", {})
|
||||
entry = list(bin_field.values())[0] if isinstance(bin_field, dict) else bin_field
|
||||
# Same priority as 9Router / MCP-bundle paths: bundled node > system node > Electron-as-Node.
|
||||
node_cmd = (bundled_node if bundled_node and os.path.exists(bundled_node) else None) \
|
||||
or shutil.which("node") \
|
||||
or electron_path
|
||||
if node_cmd:
|
||||
config["command"] = node_cmd
|
||||
config["args"] = [os.path.join(npm_dir, "node_modules", pkg_name, entry)]
|
||||
if node_cmd == electron_path:
|
||||
config.setdefault("env", {})["ELECTRON_RUN_AS_NODE"] = "1"
|
||||
logger.info(f"Using pre-installed npm MCP server for {pkg_name}")
|
||||
|
||||
if not os.path.isabs(config.get("command", "")):
|
||||
resolved = _resolve_command(config["command"])
|
||||
if resolved:
|
||||
config["command"] = resolved
|
||||
else:
|
||||
logger.warning(f"Command '{config['command']}' not found on PATH or bundled directories")
|
||||
env = config.setdefault("env", {})
|
||||
env.setdefault("PATH", _augmented_path())
|
||||
env.setdefault("PYTHONPATH", "")
|
||||
# Point uv/uvx at our bundled Python; avoids macOS CLT popup on fresh Macs
|
||||
# and avoids downloading Python at runtime
|
||||
_is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"
|
||||
_is_windows = sys.platform == "win32"
|
||||
if _is_packaged:
|
||||
_resources = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
||||
if _is_windows:
|
||||
_bundled_python = os.path.join(_resources, "python-env", "python.exe")
|
||||
else:
|
||||
_bundled_python = os.path.join(_resources, "python-env", "bin", "python3")
|
||||
if os.path.exists(_bundled_python):
|
||||
env.setdefault("UV_PYTHON", _bundled_python)
|
||||
else:
|
||||
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
if _is_windows:
|
||||
_venv_python = os.path.join(_backend, ".venv", "Scripts", "python.exe")
|
||||
else:
|
||||
_venv_python = os.path.join(_backend, ".venv", "bin", "python3")
|
||||
if os.path.exists(_venv_python):
|
||||
env.setdefault("UV_PYTHON", _venv_python)
|
||||
|
||||
return config
|
||||
@@ -0,0 +1,266 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
|
||||
import httpx
|
||||
from fastapi import HTTPException
|
||||
|
||||
from backend.apps.tools_lib.mcp_config import _augmented_path, _resolve_command
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _parse_sse_json(text: str) -> dict | None:
|
||||
"""Extract JSON from an SSE response body (handles `data: {...}` lines)."""
|
||||
for line in text.splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("data:"):
|
||||
payload = stripped[len("data:"):].strip()
|
||||
if payload:
|
||||
try:
|
||||
return json.loads(payload)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
|
||||
|
||||
async def _discover_mcp_tools_http(url: str, headers: dict | None = None) -> list[dict]:
|
||||
"""Connect to a Streamable HTTP MCP server and call tools/list via JSON-RPC POST."""
|
||||
h = {
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json, text/event-stream",
|
||||
**(headers or {}),
|
||||
}
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
init_resp = await client.post(url, headers=h, json={
|
||||
"jsonrpc": "2.0", "id": 1, "method": "initialize",
|
||||
"params": {"protocolVersion": "2025-03-26", "capabilities": {},
|
||||
"clientInfo": {"name": "self-swarm", "version": "0.1.0"}},
|
||||
})
|
||||
if init_resp.status_code not in (200, 201):
|
||||
raise HTTPException(status_code=502, detail=f"MCP initialize failed: {init_resp.status_code}")
|
||||
|
||||
session_id = init_resp.headers.get("mcp-session-id", "")
|
||||
if session_id:
|
||||
h["mcp-session-id"] = session_id
|
||||
|
||||
await client.post(url, headers=h, json={
|
||||
"jsonrpc": "2.0", "method": "notifications/initialized",
|
||||
})
|
||||
|
||||
list_resp = await client.post(url, headers=h, json={
|
||||
"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {},
|
||||
})
|
||||
if list_resp.status_code not in (200, 201):
|
||||
raise HTTPException(status_code=502, detail=f"MCP tools/list failed: {list_resp.status_code}")
|
||||
|
||||
ct = list_resp.headers.get("content-type", "")
|
||||
if "text/event-stream" in ct:
|
||||
data = _parse_sse_json(list_resp.text)
|
||||
else:
|
||||
data = list_resp.json()
|
||||
|
||||
if not data:
|
||||
raise HTTPException(status_code=502, detail="Empty response from MCP server")
|
||||
|
||||
tools_list = data.get("result", {}).get("tools", [])
|
||||
return [{"name": t.get("name", ""), "description": t.get("description", ""), "inputSchema": t.get("inputSchema")} for t in tools_list]
|
||||
|
||||
|
||||
async def _discover_mcp_tools_sse(url: str, headers: dict | None = None) -> list[dict]:
|
||||
"""Connect to a legacy SSE MCP server (GET event-stream + POST messages) and call tools/list."""
|
||||
from mcp.client.sse import sse_client
|
||||
from mcp import ClientSession
|
||||
from mcp.types import Implementation
|
||||
|
||||
try:
|
||||
async with sse_client(
|
||||
url=url,
|
||||
headers=headers,
|
||||
timeout=30,
|
||||
sse_read_timeout=30,
|
||||
) as (read_stream, write_stream):
|
||||
async with ClientSession(
|
||||
read_stream,
|
||||
write_stream,
|
||||
client_info=Implementation(name="self-swarm", version="0.1.0"),
|
||||
) as session:
|
||||
await session.initialize()
|
||||
result = await session.list_tools()
|
||||
return [{"name": t.name, "description": t.description or "", "inputSchema": t.inputSchema if t.inputSchema else None} for t in result.tools]
|
||||
except BaseExceptionGroup as eg:
|
||||
first = eg.exceptions[0] if eg.exceptions else eg
|
||||
raise HTTPException(status_code=502, detail=f"SSE discovery failed: {first}") from first
|
||||
|
||||
|
||||
_NPX_CACHE_RE = re.compile(r"_npx[/\\]([0-9a-f]{8,})[/\\]")
|
||||
|
||||
|
||||
def _try_heal_npx_cache(stderr: str) -> str | None:
|
||||
"""On `ERR_MODULE_NOT_FOUND` pointing into `~/.npm/_npx/<hash>/`, wipe that one dir.
|
||||
|
||||
Why: interrupted npx installs leave a `package-lock.json` in the cache dir so
|
||||
subsequent spawns reuse a partially-extracted node_modules tree, which dies at
|
||||
import time. Scoped strictly to the extracted hash subdir; never touches
|
||||
anything outside `~/.npm/_npx/`.
|
||||
"""
|
||||
if "ERR_MODULE_NOT_FOUND" not in stderr:
|
||||
return None
|
||||
m = _NPX_CACHE_RE.search(stderr)
|
||||
if not m:
|
||||
return None
|
||||
hash_ = m.group(1)
|
||||
cache_dir = os.path.join(os.path.expanduser("~"), ".npm", "_npx", hash_)
|
||||
if not os.path.isdir(cache_dir):
|
||||
return None
|
||||
logger.warning("Corrupted npx cache detected at %s; wiping and letting caller retry", cache_dir)
|
||||
shutil.rmtree(cache_dir, ignore_errors=True)
|
||||
return hash_
|
||||
|
||||
|
||||
async def _discover_mcp_tools_stdio(command: str, args: list[str] | None = None, env: dict | None = None, _attempt: int = 0) -> list[dict]:
|
||||
"""Spawn a stdio MCP server process and call tools/list via JSON-RPC over stdin/stdout.
|
||||
|
||||
On the first attempt, a failure that looks like corrupted npx cache
|
||||
(`ERR_MODULE_NOT_FOUND` pointing into `~/.npm/_npx/<hash>/`) triggers one
|
||||
auto-heal + retry. No heal on `_attempt >= 1`.
|
||||
"""
|
||||
cmd_path = _resolve_command(command)
|
||||
if not cmd_path:
|
||||
raise HTTPException(status_code=400, detail=f"Command '{command}' not found on PATH or common install locations")
|
||||
|
||||
proc_env = {**os.environ, **(env or {}), "PATH": _augmented_path()}
|
||||
proc_env.pop("PYTHONPATH", None)
|
||||
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
cmd_path, *(args or []),
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
env=proc_env,
|
||||
limit=10 * 1024 * 1024, # 10 MB buffer for large tool lists
|
||||
)
|
||||
|
||||
# Drain stderr in the background. Two reasons: (1) the OS pipe buffer is
|
||||
# ~64 KB; if npx prints more than that during a cold-cache install
|
||||
# (which happens when AV scanning slows npm), the child blocks on
|
||||
# write and we'd see what looks like a hang. (2) the rolling tail lets
|
||||
# us include npx's own diagnostic in any error we surface, instead of
|
||||
# the opaque "discovery failed" we used to show.
|
||||
stderr_tail: list[str] = []
|
||||
|
||||
async def _drain_stderr() -> None:
|
||||
try:
|
||||
while True:
|
||||
chunk = await proc.stderr.readline()
|
||||
if not chunk:
|
||||
return
|
||||
stderr_tail.append(chunk.decode(errors="replace"))
|
||||
if len(stderr_tail) > 50:
|
||||
del stderr_tail[: len(stderr_tail) - 50]
|
||||
except asyncio.CancelledError:
|
||||
return
|
||||
except Exception:
|
||||
return
|
||||
|
||||
stderr_task = asyncio.create_task(_drain_stderr())
|
||||
|
||||
async def _send(msg: dict) -> None:
|
||||
line = json.dumps(msg) + "\n"
|
||||
proc.stdin.write(line.encode())
|
||||
await proc.stdin.drain()
|
||||
|
||||
async def _recv(timeout_s: float = 30.0) -> dict:
|
||||
"""Read JSON-RPC responses, skipping notification lines (no 'id' field)."""
|
||||
while True:
|
||||
line = await asyncio.wait_for(proc.stdout.readline(), timeout=timeout_s)
|
||||
if not line:
|
||||
# stdout EOF = child exited. Wait briefly for the stderr
|
||||
# drain to catch up so we capture the real failure reason
|
||||
# (which often arrives a few ms after stdout closes).
|
||||
try:
|
||||
await asyncio.wait_for(asyncio.shield(stderr_task), timeout=1.0)
|
||||
except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
|
||||
pass
|
||||
tail = "".join(stderr_tail[-10:]).strip()
|
||||
raise HTTPException(
|
||||
status_code=502,
|
||||
detail=f"MCP stdio process exited unexpectedly{': ' + tail if tail else ''}",
|
||||
)
|
||||
stripped = line.decode(errors="replace").strip()
|
||||
if not stripped:
|
||||
continue
|
||||
try:
|
||||
data = json.loads(stripped)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
if "id" in data:
|
||||
return data
|
||||
|
||||
try:
|
||||
await _send({
|
||||
"jsonrpc": "2.0", "id": 1, "method": "initialize",
|
||||
"params": {
|
||||
"protocolVersion": "2025-03-26",
|
||||
"capabilities": {},
|
||||
"clientInfo": {"name": "self-swarm", "version": "0.1.0"},
|
||||
},
|
||||
})
|
||||
# First response is the slow one. On Windows with a cold npx cache,
|
||||
# `npx -y <pkg>` has to download the package + transitive deps and
|
||||
# AV-scan every file npm writes; total install time often exceeds
|
||||
# 60 s and occasionally pushes past 90 s. Subsequent reads run
|
||||
# against an already-running server and stay at the default 30 s.
|
||||
await _recv(timeout_s=120.0)
|
||||
|
||||
await _send({"jsonrpc": "2.0", "method": "notifications/initialized"})
|
||||
|
||||
await _send({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}})
|
||||
data = await _recv()
|
||||
|
||||
tools_list = data.get("result", {}).get("tools", [])
|
||||
return [{"name": t.get("name", ""), "description": t.get("description", ""), "inputSchema": t.get("inputSchema")} for t in tools_list]
|
||||
|
||||
except HTTPException as e:
|
||||
# Heal-on-corrupt-npx-cache still triggers from the EOF branch,
|
||||
# which now includes the full stderr tail in `e.detail`; so the
|
||||
# ERR_MODULE_NOT_FOUND signature is still discoverable here.
|
||||
if _attempt == 0 and _try_heal_npx_cache(str(e.detail) if e.detail is not None else ""):
|
||||
return await _discover_mcp_tools_stdio(command, args, env, _attempt=1)
|
||||
raise
|
||||
except asyncio.TimeoutError:
|
||||
# Most common cause: cold npx cache on Windows. The npm install
|
||||
# persists across attempts, so a retry usually finishes against a
|
||||
# warm cache. Surface npx's own progress line if we have one; it
|
||||
# makes the cause obvious ("downloading X...") instead of opaque.
|
||||
tail_text = "".join(stderr_tail[-5:]).strip()
|
||||
detail = "MCP discovery timed out; the server may still be downloading on first run"
|
||||
if tail_text:
|
||||
preview = tail_text[-200:].replace("\n", " ").strip()
|
||||
detail += f" (last output: {preview})"
|
||||
detail += ". Try again in a moment."
|
||||
raise HTTPException(status_code=504, detail=detail)
|
||||
finally:
|
||||
stderr_task.cancel()
|
||||
try:
|
||||
await stderr_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
try:
|
||||
proc.stdin.close()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
proc.terminate()
|
||||
await asyncio.wait_for(proc.wait(), timeout=5.0)
|
||||
except Exception:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
@@ -0,0 +1,16 @@
|
||||
import os
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from backend.config.paths import BACKEND_DIR, DATA_ROOT
|
||||
|
||||
# Loaded here (the leaf) so OPENSWARM_OAUTH_BASE_URL is set before any module
|
||||
# that reads it imports this. Both tools_lib.py and oauth_tokens.py pull from here.
|
||||
load_dotenv(os.path.join(BACKEND_DIR, ".env"))
|
||||
if os.environ.get("OPENSWARM_PACKAGED") == "1":
|
||||
load_dotenv(os.path.join(os.path.dirname(DATA_ROOT), ".env"), override=True)
|
||||
|
||||
# Base URL for the OAuth helper service. Override via env in dev if needed.
|
||||
OPENSWARM_OAUTH_BASE_URL = os.environ.get(
|
||||
"OPENSWARM_OAUTH_BASE_URL", "https://api.openswarm.com"
|
||||
).rstrip("/")
|
||||
@@ -0,0 +1,182 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
import httpx
|
||||
|
||||
from backend.config.paths import TOOLS_DIR as DATA_DIR
|
||||
from backend.apps.tools_lib.models import ToolDefinition
|
||||
from backend.apps.tools_lib.oauth_config import OPENSWARM_OAUTH_BASE_URL
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _save(tool: ToolDefinition) -> None:
|
||||
with open(os.path.join(DATA_DIR, f"{tool.id}.json"), "w") as f:
|
||||
json.dump(tool.model_dump(), f, indent=2)
|
||||
|
||||
|
||||
# Tool name → provider key for the OAuth helper service. All providers go
|
||||
# through the Fly cloud-proxy so client_secret values never ship inside the
|
||||
# desktop binary. v1.0.28 was the last release that used a local Google
|
||||
# callback with the client_secret in backend/.env.
|
||||
_TOOL_NAME_TO_PROVIDER = {
|
||||
"airtable": "airtable",
|
||||
"hubspot": "hubspot",
|
||||
"discord": "discord",
|
||||
"notion": "notion",
|
||||
# Built-in Google tool's name is "Google Workspace"; accept the bare
|
||||
# "google" alias too for forward compatibility.
|
||||
"google workspace": "google",
|
||||
"google": "google",
|
||||
}
|
||||
|
||||
|
||||
def _proxied_provider_for(tool: ToolDefinition) -> Optional[str]:
|
||||
return _TOOL_NAME_TO_PROVIDER.get(tool.name.lower())
|
||||
|
||||
|
||||
def _persist_cloud_tokens(tool: ToolDefinition, tokens: dict) -> None:
|
||||
"""Normalise the cloud's claim response into tool.oauth_tokens.
|
||||
|
||||
Per-provider shaping mirrors what the v1.0.25 local-callback flow used
|
||||
to write; the rest of the app (refresh helpers, MCP env injection)
|
||||
expects exactly this shape.
|
||||
"""
|
||||
name = tool.name.lower()
|
||||
if name == "discord":
|
||||
new_guilds = (tokens.get("_guilds") or []) if isinstance(tokens, dict) else []
|
||||
existing = tool.oauth_tokens.get("guilds") or []
|
||||
for g in new_guilds:
|
||||
if g.get("id") and not any(e.get("id") == g["id"] for e in existing):
|
||||
existing.append({"id": g["id"], "name": g.get("name", "")})
|
||||
tool.oauth_tokens = {"guilds": existing}
|
||||
names = ", ".join(g.get("name", "") for g in existing if g.get("name"))
|
||||
tool.connected_account_email = (
|
||||
f"{len(existing)} server{'s' if len(existing) != 1 else ''}"
|
||||
+ (f" · {names}" if names else "")
|
||||
)
|
||||
elif name == "notion":
|
||||
tool.oauth_tokens = {"access_token": tokens.get("access_token", "")}
|
||||
tool.connected_account_email = tokens.get("workspace_name", "Notion workspace")
|
||||
else:
|
||||
tool.oauth_tokens = {
|
||||
"access_token": tokens.get("access_token", ""),
|
||||
"refresh_token": tokens.get("refresh_token", ""),
|
||||
"token_expiry": time.time() + (tokens.get("expires_in") or 3600),
|
||||
}
|
||||
tool.connected_account_email = (
|
||||
tokens.get("email") # Google (post-userinfo enrichment)
|
||||
or tokens.get("hub_domain") # HubSpot
|
||||
or tokens.get("workspace_name")
|
||||
or f"{tool.name} account"
|
||||
)
|
||||
tool.auth_type = "oauth2"
|
||||
tool.auth_status = "connected"
|
||||
|
||||
|
||||
async def _refresh_via_proxy(provider: str, tool: ToolDefinition, default_expiry: int) -> Optional[str]:
|
||||
"""Refresh an OAuth access_token by POSTing the refresh_token to the
|
||||
helper service. Per-provider wrappers below pass a default expires_in
|
||||
fallback for providers that don't return one.
|
||||
"""
|
||||
if tool.auth_type != "oauth2":
|
||||
return None
|
||||
refresh_token = tool.oauth_tokens.get("refresh_token")
|
||||
if not refresh_token:
|
||||
return None
|
||||
expiry = tool.oauth_tokens.get("token_expiry", 0)
|
||||
if time.time() < expiry - 60:
|
||||
return tool.oauth_tokens.get("access_token")
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||||
resp = await client.post(
|
||||
f"{OPENSWARM_OAUTH_BASE_URL}/api/oauth/{provider}/refresh",
|
||||
json={"refresh_token": refresh_token},
|
||||
)
|
||||
if resp.status_code == 401:
|
||||
# Provider rejected; user revoked at the provider's side. Mark
|
||||
# as needing re-auth so the UI prompts a Reconnect.
|
||||
tool.auth_status = "expired"
|
||||
_save(tool)
|
||||
logger.warning(f"{provider} refresh rejected (user revoked); marking tool as expired")
|
||||
return None
|
||||
if resp.status_code != 200:
|
||||
logger.warning(f"{provider} cloud refresh failed: HTTP %d %s", resp.status_code, resp.text[:200])
|
||||
return None
|
||||
|
||||
data = (resp.json() or {}).get("tokens") or {}
|
||||
new_token = data.get("access_token", "")
|
||||
if not new_token:
|
||||
return None
|
||||
tool.oauth_tokens["access_token"] = new_token
|
||||
tool.oauth_tokens["token_expiry"] = time.time() + (data.get("expires_in") or default_expiry)
|
||||
if data.get("refresh_token"):
|
||||
# Some providers (HubSpot, Airtable) rotate refresh_tokens on every
|
||||
# refresh. Persist the new one or future refreshes will fail.
|
||||
tool.oauth_tokens["refresh_token"] = data["refresh_token"]
|
||||
# Backfill identity label on first successful refresh after upgrade.
|
||||
if not tool.connected_account_email and data.get("email"):
|
||||
tool.connected_account_email = data["email"]
|
||||
_save(tool)
|
||||
return new_token
|
||||
except Exception as e:
|
||||
logger.warning(f"{provider} cloud refresh exception for tool {tool.id}: {e}")
|
||||
return None
|
||||
|
||||
|
||||
async def refresh_google_token(tool: ToolDefinition) -> Optional[str]:
|
||||
"""Refresh an expired Google access_token via the Fly cloud-proxy.
|
||||
|
||||
The client_secret never leaves Fly; desktop only POSTs the
|
||||
refresh_token. Same pattern as Airtable/HubSpot. Pre-v1.0.29 builds
|
||||
held the secret in their bundled .env; v1.0.29 removed it.
|
||||
"""
|
||||
return await _refresh_via_proxy("google", tool, default_expiry=3600)
|
||||
|
||||
|
||||
async def refresh_airtable_token(tool: ToolDefinition) -> Optional[str]:
|
||||
"""Refresh an expired Airtable OAuth access_token."""
|
||||
return await _refresh_via_proxy("airtable", tool, default_expiry=7200)
|
||||
|
||||
|
||||
async def refresh_hubspot_token(tool: ToolDefinition) -> Optional[str]:
|
||||
"""Refresh an expired HubSpot OAuth access_token."""
|
||||
return await _refresh_via_proxy("hubspot", tool, default_expiry=1800)
|
||||
|
||||
|
||||
def _m365_server_script() -> str:
|
||||
"""Return the on-disk path to the bundled MS365 MCP server entry.
|
||||
|
||||
v1.0.26 replaced the heavy backend/npm-servers/softeria-ms-365-mcp-server/
|
||||
node_modules tree (~93MB / 11k files) with a single esbuild bundle at
|
||||
backend/mcp-bundles/softeria-ms-365-mcp-server/dist/index.js (4.7MB).
|
||||
The new path mirrors the SDK's internal layout (dist/index.js + sibling
|
||||
package.json) because cli.js reads __dirname/../package.json for the
|
||||
--version flag; see scripts/build-app.sh `build_mcp_bundle_dir`.
|
||||
"""
|
||||
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
bundle = os.path.join(
|
||||
_backend, "mcp-bundles", "softeria-ms-365-mcp-server", "dist", "index.js",
|
||||
)
|
||||
if os.path.isfile(bundle):
|
||||
return bundle
|
||||
# Fallback for any user still on a v1.0.25 install whose backend/ folder
|
||||
# was left over from before the bundle migration. Will return the legacy
|
||||
# path; if that doesn't exist either, the caller raises a clear error.
|
||||
return os.path.join(
|
||||
_backend, "npm-servers", "softeria-ms-365-mcp-server",
|
||||
"node_modules", "@softeria", "ms-365-mcp-server", "dist", "index.js",
|
||||
)
|
||||
|
||||
|
||||
def _m365_cache_env() -> dict[str, str]:
|
||||
cache_dir = os.path.join(os.path.expanduser("~"), ".openswarm")
|
||||
os.makedirs(cache_dir, exist_ok=True)
|
||||
return {
|
||||
"MS365_MCP_TOKEN_CACHE_PATH": os.path.join(cache_dir, "ms365-token-cache.json"),
|
||||
"MS365_MCP_SELECTED_ACCOUNT_PATH": os.path.join(cache_dir, "ms365-selected-account.json"),
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
_READ_PREFIXES = ("get", "list", "read", "search", "fetch", "find", "query", "count", "check", "describe", "show", "download", "browse", "analy", "explain")
|
||||
_WRITE_PREFIXES = ("create", "write", "delete", "update", "send", "remove", "modify", "add", "set", "put", "post", "patch", "insert", "move", "copy", "rename", "archive", "trash", "publish", "approve", "reject")
|
||||
|
||||
|
||||
_SERVICE_RULES: list[tuple[list[str], str, str]] = [
|
||||
# (keywords, service_name, group)
|
||||
# Google Workspace
|
||||
(["gmail"], "Gmail", "Google"),
|
||||
(["drive"], "Drive", "Google"),
|
||||
(["calendar", "event", "freebusy"], "Calendar", "Google"),
|
||||
(["spreadsheet", "sheet"], "Sheets", "Google"),
|
||||
(["doc", "paragraph", "table"], "Docs", "Google"),
|
||||
(["chat", "space", "reaction", "message"], "Chat", "Google"),
|
||||
(["form", "publish_settings"], "Forms", "Google"),
|
||||
(["presentation", "slide", "page"], "Slides", "Google"),
|
||||
(["task_list", "task"], "Tasks", "Google"),
|
||||
(["contact"], "Contacts", "Google"),
|
||||
(["script", "deployment", "version", "trigger"], "Apps Script", "Google"),
|
||||
(["search_custom", "search_engine"], "Search", "Google"),
|
||||
# YouTube
|
||||
(["transcript", "caption"], "Transcripts", "YouTube"),
|
||||
(["video_detail", "video_comment", "video_categor", "video_engagement"], "Videos", "YouTube"),
|
||||
(["search_video", "trending_video"], "Search", "YouTube"),
|
||||
(["channel_stat", "channel_top"], "Channels", "YouTube"),
|
||||
# Reddit (before Twitter so "search_reddit" etc. don't mis-match)
|
||||
(["subreddit"], "Subreddits", "Reddit"),
|
||||
(["search_reddit"], "Search", "Reddit"),
|
||||
(["post_detail"], "Posts", "Reddit"),
|
||||
(["user_analysis"], "Users", "Reddit"),
|
||||
(["reddit_explain"], "Reference", "Reddit"),
|
||||
]
|
||||
|
||||
|
||||
def _categorize_tool(name: str) -> str:
|
||||
lower = name.lower().replace("_", " ").replace("-", " ").strip()
|
||||
for word in lower.split():
|
||||
for prefix in _READ_PREFIXES:
|
||||
if word.startswith(prefix):
|
||||
return "read"
|
||||
for prefix in _WRITE_PREFIXES:
|
||||
if word.startswith(prefix):
|
||||
return "write"
|
||||
return "write"
|
||||
|
||||
|
||||
def _extract_service(name: str) -> tuple[str, str]:
|
||||
"""Extract the service and group from a tool name (e.g. 'search_gmail_messages' -> ('Gmail', 'Google'))."""
|
||||
lower = name.lower()
|
||||
for keywords, display, group in _SERVICE_RULES:
|
||||
for kw in keywords:
|
||||
if kw in lower:
|
||||
return display, group
|
||||
return "Other", ""
|
||||
@@ -1,37 +1,44 @@
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import time
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import Any, Optional
|
||||
from typing import Any
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import httpx
|
||||
from dotenv import load_dotenv
|
||||
from fastapi import HTTPException, Query, Request, Response
|
||||
from fastapi.responses import HTMLResponse
|
||||
from pydantic import BaseModel
|
||||
from backend.config.Apps import SubApp
|
||||
from backend.apps.tools_lib.models import ToolDefinition, ToolCreate, ToolUpdate, BUILTIN_TOOLS
|
||||
from backend.config.paths import DATA_ROOT, TOOLS_DIR as DATA_DIR, BUILTIN_PERMISSIONS_PATH as BUILTIN_PERMS_PATH, TRUSTED_SENSITIVE_PATHS_PATH
|
||||
|
||||
# oauth_config runs the dotenv load (leaf) so OPENSWARM_OAUTH_BASE_URL is set
|
||||
# before anything reads it; re-exported here for the route handlers below.
|
||||
from backend.apps.tools_lib.oauth_config import OPENSWARM_OAUTH_BASE_URL
|
||||
# _sanitize_server_name + derive_mcp_config re-exported for agent_manager/main.
|
||||
from backend.apps.tools_lib.mcp_config import _sanitize_server_name, derive_mcp_config
|
||||
from backend.apps.tools_lib.mcp_discovery import (
|
||||
_discover_mcp_tools_http,
|
||||
_discover_mcp_tools_sse,
|
||||
_discover_mcp_tools_stdio,
|
||||
)
|
||||
from backend.apps.tools_lib.tool_taxonomy import _categorize_tool, _extract_service
|
||||
# refresh_* re-exported for agent_manager.
|
||||
from backend.apps.tools_lib.oauth_tokens import (
|
||||
_proxied_provider_for,
|
||||
_persist_cloud_tokens,
|
||||
refresh_google_token,
|
||||
refresh_airtable_token,
|
||||
refresh_hubspot_token,
|
||||
_m365_server_script,
|
||||
_m365_cache_env,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Base URL for the OAuth helper service. Override via env in dev if needed.
|
||||
OPENSWARM_OAUTH_BASE_URL = os.environ.get(
|
||||
"OPENSWARM_OAUTH_BASE_URL", "https://api.openswarm.com"
|
||||
).rstrip("/")
|
||||
|
||||
from backend.config.paths import BACKEND_DIR, DATA_ROOT, TOOLS_DIR as DATA_DIR, BUILTIN_PERMISSIONS_PATH as BUILTIN_PERMS_PATH, TRUSTED_SENSITIVE_PATHS_PATH
|
||||
|
||||
load_dotenv(os.path.join(BACKEND_DIR, ".env"))
|
||||
if os.environ.get("OPENSWARM_PACKAGED") == "1":
|
||||
load_dotenv(os.path.join(os.path.dirname(DATA_ROOT), ".env"), override=True)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def tools_lib_lifespan():
|
||||
@@ -252,610 +259,6 @@ async def delete_tool(tool_id: str):
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MCP config derivation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _sanitize_server_name(name: str) -> str:
|
||||
"""Convert a tool name into a valid MCP server identifier (alphanumeric + hyphens)."""
|
||||
return re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
|
||||
|
||||
|
||||
def _extra_bin_dirs() -> list[str]:
|
||||
"""Well-known user-local bin directories that may not be on PATH in packaged apps."""
|
||||
home = os.path.expanduser("~")
|
||||
# Bundled uv-bin (ships uvx for non-dev users)
|
||||
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
dirs = [
|
||||
os.path.join(_backend, "uv-bin"),
|
||||
os.path.join(home, ".bun", "bin"),
|
||||
os.path.join(home, ".cargo", "bin"),
|
||||
os.path.join(home, ".local", "bin"),
|
||||
os.path.join(home, ".volta", "bin"),
|
||||
"/opt/homebrew/bin",
|
||||
"/usr/local/bin",
|
||||
]
|
||||
# nvm: pick the newest installed node version
|
||||
nvm_node = os.path.join(home, ".nvm", "versions", "node")
|
||||
try:
|
||||
if os.path.isdir(nvm_node):
|
||||
versions = sorted(os.listdir(nvm_node), reverse=True)
|
||||
if versions:
|
||||
dirs.insert(0, os.path.join(nvm_node, versions[0], "bin"))
|
||||
except OSError:
|
||||
pass
|
||||
# fnm
|
||||
fnm_bin = os.path.join(home, "Library", "Application Support", "fnm", "aliases", "default", "bin")
|
||||
if os.path.isdir(fnm_bin):
|
||||
dirs.insert(0, fnm_bin)
|
||||
return dirs
|
||||
|
||||
|
||||
def _resolve_command(command: str) -> str | None:
|
||||
"""Find a command on PATH, falling back to common user-local bin directories
|
||||
and bundled binaries (uv-bin for uvx/uv)."""
|
||||
found = shutil.which(command)
|
||||
if found:
|
||||
return found
|
||||
# Windows binaries need an extension. shutil.which() handles PATHEXT for
|
||||
# PATH lookups, but we manually scan _extra_bin_dirs below; replicate
|
||||
# the suffix probing here so `uvx` finds `uvx.exe`, etc.
|
||||
if sys.platform == "win32":
|
||||
suffixes = [""] + os.environ.get("PATHEXT", ".COM;.EXE;.BAT;.CMD").lower().split(os.pathsep)
|
||||
else:
|
||||
suffixes = [""]
|
||||
def _probe(directory: str) -> str | None:
|
||||
for suffix in suffixes:
|
||||
candidate = os.path.join(directory, command + suffix)
|
||||
if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
|
||||
return candidate
|
||||
return None
|
||||
for d in _extra_bin_dirs():
|
||||
hit = _probe(d)
|
||||
if hit:
|
||||
return hit
|
||||
# Check bundled uv-bin directory (ships uv/uvx for non-dev users)
|
||||
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
return _probe(os.path.join(_backend, "uv-bin"))
|
||||
|
||||
|
||||
def _augmented_path() -> str:
|
||||
"""Return PATH with extra bin dirs prepended (for child process environments)."""
|
||||
extra = [d for d in _extra_bin_dirs() if os.path.isdir(d)]
|
||||
current = os.environ.get("PATH", "")
|
||||
seen: set[str] = set()
|
||||
parts: list[str] = []
|
||||
for p in extra + current.split(os.pathsep):
|
||||
if p and p not in seen:
|
||||
seen.add(p)
|
||||
parts.append(p)
|
||||
return os.pathsep.join(parts)
|
||||
|
||||
|
||||
def derive_mcp_config(tool: ToolDefinition) -> Optional[dict]:
|
||||
"""Build the claude_agent_sdk mcp_servers config entry for a tool.
|
||||
|
||||
Returns None if the tool cannot be configured (e.g. missing data).
|
||||
"""
|
||||
if not tool.mcp_config:
|
||||
return None
|
||||
|
||||
config: dict = dict(tool.mcp_config)
|
||||
|
||||
if tool.credentials:
|
||||
if config.get("type") in ("http", "sse"):
|
||||
headers = config.setdefault("headers", {})
|
||||
for key, val in tool.credentials.items():
|
||||
if key.lower() in ("authorization", "api_key", "api-key"):
|
||||
headers.setdefault("Authorization", f"Bearer {val}")
|
||||
else:
|
||||
env = config.setdefault("env", {})
|
||||
env.update(tool.credentials)
|
||||
|
||||
if tool.oauth_tokens.get("access_token"):
|
||||
if config.get("type") in ("http", "sse"):
|
||||
headers = config.setdefault("headers", {})
|
||||
headers["Authorization"] = f"Bearer {tool.oauth_tokens['access_token']}"
|
||||
else:
|
||||
env = config.setdefault("env", {})
|
||||
env["OAUTH_ACCESS_TOKEN"] = tool.oauth_tokens["access_token"]
|
||||
if tool.name.lower() == "notion":
|
||||
env["NOTION_TOKEN"] = tool.oauth_tokens["access_token"]
|
||||
if tool.name.lower() == "hubspot":
|
||||
env["PRIVATE_APP_ACCESS_TOKEN"] = tool.oauth_tokens["access_token"]
|
||||
if tool.oauth_tokens.get("refresh_token"):
|
||||
env["GOOGLE_WORKSPACE_REFRESH_TOKEN"] = tool.oauth_tokens["refresh_token"]
|
||||
# google_workspace_mcp's gauth.py hardcodes token_uri to
|
||||
# https://oauth2.googleapis.com/token and refreshes using the
|
||||
# local CLIENT_ID/SECRET on every API call. The OAuth flow
|
||||
# itself runs through the cloud's rotation pool, so the
|
||||
# refresh_token is bound to whichever pool slot minted it,
|
||||
# not the single client baked into the DMG. Mismatch -> Google
|
||||
# returns unauthorized_client. We point token_uri at a local
|
||||
# proxy that forwards the refresh to our cloud's pool-aware
|
||||
# /api/oauth/google/refresh endpoint; CLIENT_ID/SECRET become
|
||||
# unused placeholders (gauth.py only validates non-empty).
|
||||
_port = os.environ.get("OPENSWARM_PORT", "8324")
|
||||
env["GOOGLE_WORKSPACE_TOKEN_URI"] = (
|
||||
f"http://127.0.0.1:{_port}/api/tools/google-oauth-token"
|
||||
)
|
||||
env.setdefault("GOOGLE_WORKSPACE_CLIENT_ID", "openswarm-proxy")
|
||||
env.setdefault("GOOGLE_WORKSPACE_CLIENT_SECRET", "openswarm-proxy")
|
||||
|
||||
# Google Workspace MCP: redirect spawn through our shim that
|
||||
# monkey-patches gauth.get_credentials before the worker registers
|
||||
# tools, so token_uri points at our local proxy. Stays a stdio
|
||||
# subprocess; google-workspace-mcp gets installed into uv's
|
||||
# ephemeral env via --with, same way the upstream entry-point
|
||||
# invocation used to do it.
|
||||
if tool.name.lower() == "google workspace" and config.get("type") == "stdio":
|
||||
shim_path = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
"google_workspace_mcp_shim",
|
||||
"run.py",
|
||||
)
|
||||
config["command"] = "uv"
|
||||
config["args"] = ["run", "--with", "google-workspace-mcp", "python", shim_path]
|
||||
|
||||
# Discord MCP runs as a small Python shim (backend.apps.discord_mcp_shim).
|
||||
# We pass install_id + base URL via env so the shim subprocess doesn't
|
||||
# need to import backend.config.* itself.
|
||||
if tool.name.lower() == "discord" and config.get("type") == "stdio":
|
||||
from backend.config.install_id import get_install_id
|
||||
env = config.setdefault("env", {})
|
||||
env["OPENSWARM_OAUTH_BASE_URL"] = OPENSWARM_OAUTH_BASE_URL
|
||||
env["OPENSWARM_INSTALL_ID"] = get_install_id()
|
||||
# Pass the authorized guild IDs so the shim can scope-enforce.
|
||||
guild_ids = [g.get("id", "") for g in (tool.oauth_tokens.get("guilds") or []) if g.get("id")]
|
||||
if guild_ids:
|
||||
env["OPENSWARM_DISCORD_GUILD_IDS"] = ",".join(guild_ids)
|
||||
# The shim runs as a subprocess and needs to import
|
||||
# `backend.apps.discord_mcp_shim`; set PYTHONPATH to the project
|
||||
# root (parent of the backend/ dir) so that import resolves.
|
||||
_project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
||||
existing_pp = env.get("PYTHONPATH") or os.environ.get("PYTHONPATH", "")
|
||||
env["PYTHONPATH"] = (_project_root + os.pathsep + existing_pp) if existing_pp else _project_root
|
||||
|
||||
# Microsoft 365 MCP: use a stable token cache path shared across process spawns
|
||||
if tool.name.lower() == "microsoft 365" and config.get("type") == "stdio":
|
||||
env = config.setdefault("env", {})
|
||||
cache_dir = os.path.join(os.path.expanduser("~"), ".openswarm")
|
||||
os.makedirs(cache_dir, exist_ok=True)
|
||||
env["MS365_MCP_TOKEN_CACHE_PATH"] = os.path.join(cache_dir, "ms365-token-cache.json")
|
||||
env["MS365_MCP_SELECTED_ACCOUNT_PATH"] = os.path.join(cache_dir, "ms365-selected-account.json")
|
||||
|
||||
if config.get("type") == "stdio":
|
||||
if config.get("command"):
|
||||
# `python` (no version suffix) doesn't exist on a stock macOS,
|
||||
# so a tool config that asks for "python" silently fails to
|
||||
# spawn; Claude Agent SDK then exposes zero tools from that
|
||||
# MCP. We resolve to the actual interpreter running the
|
||||
# backend (sys.executable), which is guaranteed to exist and
|
||||
# have backend modules importable. `python3` and absolute
|
||||
# paths pass through unchanged.
|
||||
if config["command"] == "python":
|
||||
resolved_python = sys.executable or shutil.which("python3") or shutil.which("python")
|
||||
if resolved_python:
|
||||
config["command"] = resolved_python
|
||||
# Check for bundled npm MCP servers; use Electron's Node.js instead of npx
|
||||
if config["command"] in ("npx", "bunx"):
|
||||
pkg_name = next((a for a in (config.get("args") or []) if not a.startswith("-")), None)
|
||||
if pkg_name:
|
||||
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
electron_path = os.environ.get("OPENSWARM_ELECTRON_PATH")
|
||||
# Two bundle layouts in mcp-bundles/, checked in priority order:
|
||||
#
|
||||
# 1. Multi-file bundle dir: mcp-bundles/<safe>/dist/index.js
|
||||
# Used when the SDK reads sibling files at runtime.
|
||||
# Examples: @softeria/ms-365-mcp-server reads
|
||||
# ../package.json for --version and dist/endpoints.json
|
||||
# for Graph API definitions; @notionhq/notion-mcp-server
|
||||
# reads ../scripts/notion-openapi.json. The build script
|
||||
# ships a stripped package.json (no "type":"module") next
|
||||
# to dist/ so __dirname/../package.json resolves correctly.
|
||||
# See scripts/build-app.sh `build_mcp_bundle_dir`.
|
||||
#
|
||||
# 2. Single-file bundle: mcp-bundles/<safe>.js
|
||||
# Used when the SDK is fully self-contained
|
||||
# (reddit-mcp-buddy).
|
||||
#
|
||||
# Scoped names get flattened ("@softeria/ms-365-mcp-server"
|
||||
# -> "softeria-ms-365-mcp-server") for filesystem safety.
|
||||
safe_bundle = pkg_name.replace("/", "-").replace("@", "")
|
||||
bundle_dir_path = os.path.join(_backend, "mcp-bundles", safe_bundle, "dist", "index.js")
|
||||
bundle_file_path = os.path.join(_backend, "mcp-bundles", f"{safe_bundle}.js")
|
||||
bundle_path = None
|
||||
if os.path.isfile(bundle_dir_path):
|
||||
bundle_path = bundle_dir_path
|
||||
elif os.path.isfile(bundle_file_path):
|
||||
bundle_path = bundle_file_path
|
||||
# Prefer the bundled real-Node binary over Electron-as-Node:
|
||||
# avoids the bouncing "exec" Dock icon on fresh user Macs +
|
||||
# spawns ~10x faster than re-execing the OpenSwarm Electron
|
||||
# binary as Node. Falls back to Electron-as-Node only if
|
||||
# the bundled node payload wasn't shipped (legacy builds).
|
||||
bundled_node = os.environ.get("OPENSWARM_NODE_PATH")
|
||||
if bundle_path and bundled_node and os.path.exists(bundled_node):
|
||||
config["command"] = bundled_node
|
||||
config["args"] = [bundle_path]
|
||||
logger.info(f"Using bundled MCP server for {pkg_name} via bundled node ({bundle_path})")
|
||||
elif bundle_path and electron_path:
|
||||
config["command"] = electron_path
|
||||
config["args"] = [bundle_path]
|
||||
config.setdefault("env", {})["ELECTRON_RUN_AS_NODE"] = "1"
|
||||
logger.info(f"Using bundled MCP server for {pkg_name} ({bundle_path})")
|
||||
else:
|
||||
# Check for pre-installed npm package (works in both dev and packaged modes)
|
||||
safe_dir = pkg_name.replace("/", "-").replace("@", "")
|
||||
npm_dir = os.path.join(_backend, "npm-servers", safe_dir)
|
||||
pkg_json_path = os.path.join(npm_dir, "node_modules", pkg_name, "package.json")
|
||||
if os.path.isfile(pkg_json_path):
|
||||
import json as _json
|
||||
with open(pkg_json_path) as f:
|
||||
pkg_meta = _json.load(f)
|
||||
bin_field = pkg_meta.get("bin", {})
|
||||
entry = list(bin_field.values())[0] if isinstance(bin_field, dict) else bin_field
|
||||
# Same priority as 9Router / MCP-bundle paths: bundled node > system node > Electron-as-Node.
|
||||
node_cmd = (bundled_node if bundled_node and os.path.exists(bundled_node) else None) \
|
||||
or shutil.which("node") \
|
||||
or electron_path
|
||||
if node_cmd:
|
||||
config["command"] = node_cmd
|
||||
config["args"] = [os.path.join(npm_dir, "node_modules", pkg_name, entry)]
|
||||
if node_cmd == electron_path:
|
||||
config.setdefault("env", {})["ELECTRON_RUN_AS_NODE"] = "1"
|
||||
logger.info(f"Using pre-installed npm MCP server for {pkg_name}")
|
||||
|
||||
if not os.path.isabs(config.get("command", "")):
|
||||
resolved = _resolve_command(config["command"])
|
||||
if resolved:
|
||||
config["command"] = resolved
|
||||
else:
|
||||
logger.warning(f"Command '{config['command']}' not found on PATH or bundled directories")
|
||||
env = config.setdefault("env", {})
|
||||
env.setdefault("PATH", _augmented_path())
|
||||
env.setdefault("PYTHONPATH", "")
|
||||
# Point uv/uvx at our bundled Python; avoids macOS CLT popup on fresh Macs
|
||||
# and avoids downloading Python at runtime
|
||||
_is_packaged = os.environ.get("OPENSWARM_PACKAGED") == "1"
|
||||
_is_windows = sys.platform == "win32"
|
||||
if _is_packaged:
|
||||
_resources = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
||||
if _is_windows:
|
||||
_bundled_python = os.path.join(_resources, "python-env", "python.exe")
|
||||
else:
|
||||
_bundled_python = os.path.join(_resources, "python-env", "bin", "python3")
|
||||
if os.path.exists(_bundled_python):
|
||||
env.setdefault("UV_PYTHON", _bundled_python)
|
||||
else:
|
||||
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
if _is_windows:
|
||||
_venv_python = os.path.join(_backend, ".venv", "Scripts", "python.exe")
|
||||
else:
|
||||
_venv_python = os.path.join(_backend, ".venv", "bin", "python3")
|
||||
if os.path.exists(_venv_python):
|
||||
env.setdefault("UV_PYTHON", _venv_python)
|
||||
|
||||
return config
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# OAuth2 flow for Google Workspace (and other OAuth providers)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MCP tool discovery
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_READ_PREFIXES = ("get", "list", "read", "search", "fetch", "find", "query", "count", "check", "describe", "show", "download", "browse", "analy", "explain")
|
||||
_WRITE_PREFIXES = ("create", "write", "delete", "update", "send", "remove", "modify", "add", "set", "put", "post", "patch", "insert", "move", "copy", "rename", "archive", "trash", "publish", "approve", "reject")
|
||||
|
||||
|
||||
_SERVICE_RULES: list[tuple[list[str], str, str]] = [
|
||||
# (keywords, service_name, group)
|
||||
# Google Workspace
|
||||
(["gmail"], "Gmail", "Google"),
|
||||
(["drive"], "Drive", "Google"),
|
||||
(["calendar", "event", "freebusy"], "Calendar", "Google"),
|
||||
(["spreadsheet", "sheet"], "Sheets", "Google"),
|
||||
(["doc", "paragraph", "table"], "Docs", "Google"),
|
||||
(["chat", "space", "reaction", "message"], "Chat", "Google"),
|
||||
(["form", "publish_settings"], "Forms", "Google"),
|
||||
(["presentation", "slide", "page"], "Slides", "Google"),
|
||||
(["task_list", "task"], "Tasks", "Google"),
|
||||
(["contact"], "Contacts", "Google"),
|
||||
(["script", "deployment", "version", "trigger"], "Apps Script", "Google"),
|
||||
(["search_custom", "search_engine"], "Search", "Google"),
|
||||
# YouTube
|
||||
(["transcript", "caption"], "Transcripts", "YouTube"),
|
||||
(["video_detail", "video_comment", "video_categor", "video_engagement"], "Videos", "YouTube"),
|
||||
(["search_video", "trending_video"], "Search", "YouTube"),
|
||||
(["channel_stat", "channel_top"], "Channels", "YouTube"),
|
||||
# Reddit (before Twitter so "search_reddit" etc. don't mis-match)
|
||||
(["subreddit"], "Subreddits", "Reddit"),
|
||||
(["search_reddit"], "Search", "Reddit"),
|
||||
(["post_detail"], "Posts", "Reddit"),
|
||||
(["user_analysis"], "Users", "Reddit"),
|
||||
(["reddit_explain"], "Reference", "Reddit"),
|
||||
]
|
||||
|
||||
|
||||
def _categorize_tool(name: str) -> str:
|
||||
lower = name.lower().replace("_", " ").replace("-", " ").strip()
|
||||
for word in lower.split():
|
||||
for prefix in _READ_PREFIXES:
|
||||
if word.startswith(prefix):
|
||||
return "read"
|
||||
for prefix in _WRITE_PREFIXES:
|
||||
if word.startswith(prefix):
|
||||
return "write"
|
||||
return "write"
|
||||
|
||||
|
||||
def _extract_service(name: str) -> tuple[str, str]:
|
||||
"""Extract the service and group from a tool name (e.g. 'search_gmail_messages' -> ('Gmail', 'Google'))."""
|
||||
lower = name.lower()
|
||||
for keywords, display, group in _SERVICE_RULES:
|
||||
for kw in keywords:
|
||||
if kw in lower:
|
||||
return display, group
|
||||
return "Other", ""
|
||||
|
||||
|
||||
def _parse_sse_json(text: str) -> dict | None:
|
||||
"""Extract JSON from an SSE response body (handles `data: {...}` lines)."""
|
||||
for line in text.splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("data:"):
|
||||
payload = stripped[len("data:"):].strip()
|
||||
if payload:
|
||||
try:
|
||||
return json.loads(payload)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
|
||||
|
||||
async def _discover_mcp_tools_http(url: str, headers: dict | None = None) -> list[dict]:
|
||||
"""Connect to a Streamable HTTP MCP server and call tools/list via JSON-RPC POST."""
|
||||
h = {
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json, text/event-stream",
|
||||
**(headers or {}),
|
||||
}
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
init_resp = await client.post(url, headers=h, json={
|
||||
"jsonrpc": "2.0", "id": 1, "method": "initialize",
|
||||
"params": {"protocolVersion": "2025-03-26", "capabilities": {},
|
||||
"clientInfo": {"name": "self-swarm", "version": "0.1.0"}},
|
||||
})
|
||||
if init_resp.status_code not in (200, 201):
|
||||
raise HTTPException(status_code=502, detail=f"MCP initialize failed: {init_resp.status_code}")
|
||||
|
||||
session_id = init_resp.headers.get("mcp-session-id", "")
|
||||
if session_id:
|
||||
h["mcp-session-id"] = session_id
|
||||
|
||||
await client.post(url, headers=h, json={
|
||||
"jsonrpc": "2.0", "method": "notifications/initialized",
|
||||
})
|
||||
|
||||
list_resp = await client.post(url, headers=h, json={
|
||||
"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {},
|
||||
})
|
||||
if list_resp.status_code not in (200, 201):
|
||||
raise HTTPException(status_code=502, detail=f"MCP tools/list failed: {list_resp.status_code}")
|
||||
|
||||
ct = list_resp.headers.get("content-type", "")
|
||||
if "text/event-stream" in ct:
|
||||
data = _parse_sse_json(list_resp.text)
|
||||
else:
|
||||
data = list_resp.json()
|
||||
|
||||
if not data:
|
||||
raise HTTPException(status_code=502, detail="Empty response from MCP server")
|
||||
|
||||
tools_list = data.get("result", {}).get("tools", [])
|
||||
return [{"name": t.get("name", ""), "description": t.get("description", ""), "inputSchema": t.get("inputSchema")} for t in tools_list]
|
||||
|
||||
|
||||
async def _discover_mcp_tools_sse(url: str, headers: dict | None = None) -> list[dict]:
|
||||
"""Connect to a legacy SSE MCP server (GET event-stream + POST messages) and call tools/list."""
|
||||
from mcp.client.sse import sse_client
|
||||
from mcp import ClientSession
|
||||
from mcp.types import Implementation
|
||||
|
||||
try:
|
||||
async with sse_client(
|
||||
url=url,
|
||||
headers=headers,
|
||||
timeout=30,
|
||||
sse_read_timeout=30,
|
||||
) as (read_stream, write_stream):
|
||||
async with ClientSession(
|
||||
read_stream,
|
||||
write_stream,
|
||||
client_info=Implementation(name="self-swarm", version="0.1.0"),
|
||||
) as session:
|
||||
await session.initialize()
|
||||
result = await session.list_tools()
|
||||
return [{"name": t.name, "description": t.description or "", "inputSchema": t.inputSchema if t.inputSchema else None} for t in result.tools]
|
||||
except BaseExceptionGroup as eg:
|
||||
first = eg.exceptions[0] if eg.exceptions else eg
|
||||
raise HTTPException(status_code=502, detail=f"SSE discovery failed: {first}") from first
|
||||
|
||||
|
||||
_NPX_CACHE_RE = re.compile(r"_npx[/\\]([0-9a-f]{8,})[/\\]")
|
||||
|
||||
|
||||
def _try_heal_npx_cache(stderr: str) -> str | None:
|
||||
"""On `ERR_MODULE_NOT_FOUND` pointing into `~/.npm/_npx/<hash>/`, wipe that one dir.
|
||||
|
||||
Why: interrupted npx installs leave a `package-lock.json` in the cache dir so
|
||||
subsequent spawns reuse a partially-extracted node_modules tree, which dies at
|
||||
import time. Scoped strictly to the extracted hash subdir; never touches
|
||||
anything outside `~/.npm/_npx/`.
|
||||
"""
|
||||
if "ERR_MODULE_NOT_FOUND" not in stderr:
|
||||
return None
|
||||
m = _NPX_CACHE_RE.search(stderr)
|
||||
if not m:
|
||||
return None
|
||||
hash_ = m.group(1)
|
||||
cache_dir = os.path.join(os.path.expanduser("~"), ".npm", "_npx", hash_)
|
||||
if not os.path.isdir(cache_dir):
|
||||
return None
|
||||
logger.warning("Corrupted npx cache detected at %s; wiping and letting caller retry", cache_dir)
|
||||
shutil.rmtree(cache_dir, ignore_errors=True)
|
||||
return hash_
|
||||
|
||||
|
||||
async def _discover_mcp_tools_stdio(command: str, args: list[str] | None = None, env: dict | None = None, _attempt: int = 0) -> list[dict]:
|
||||
"""Spawn a stdio MCP server process and call tools/list via JSON-RPC over stdin/stdout.
|
||||
|
||||
On the first attempt, a failure that looks like corrupted npx cache
|
||||
(`ERR_MODULE_NOT_FOUND` pointing into `~/.npm/_npx/<hash>/`) triggers one
|
||||
auto-heal + retry. No heal on `_attempt >= 1`.
|
||||
"""
|
||||
cmd_path = _resolve_command(command)
|
||||
if not cmd_path:
|
||||
raise HTTPException(status_code=400, detail=f"Command '{command}' not found on PATH or common install locations")
|
||||
|
||||
proc_env = {**os.environ, **(env or {}), "PATH": _augmented_path()}
|
||||
proc_env.pop("PYTHONPATH", None)
|
||||
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
cmd_path, *(args or []),
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
env=proc_env,
|
||||
limit=10 * 1024 * 1024, # 10 MB buffer for large tool lists
|
||||
)
|
||||
|
||||
# Drain stderr in the background. Two reasons: (1) the OS pipe buffer is
|
||||
# ~64 KB; if npx prints more than that during a cold-cache install
|
||||
# (which happens when AV scanning slows npm), the child blocks on
|
||||
# write and we'd see what looks like a hang. (2) the rolling tail lets
|
||||
# us include npx's own diagnostic in any error we surface, instead of
|
||||
# the opaque "discovery failed" we used to show.
|
||||
stderr_tail: list[str] = []
|
||||
|
||||
async def _drain_stderr() -> None:
|
||||
try:
|
||||
while True:
|
||||
chunk = await proc.stderr.readline()
|
||||
if not chunk:
|
||||
return
|
||||
stderr_tail.append(chunk.decode(errors="replace"))
|
||||
if len(stderr_tail) > 50:
|
||||
del stderr_tail[: len(stderr_tail) - 50]
|
||||
except asyncio.CancelledError:
|
||||
return
|
||||
except Exception:
|
||||
return
|
||||
|
||||
stderr_task = asyncio.create_task(_drain_stderr())
|
||||
|
||||
async def _send(msg: dict) -> None:
|
||||
line = json.dumps(msg) + "\n"
|
||||
proc.stdin.write(line.encode())
|
||||
await proc.stdin.drain()
|
||||
|
||||
async def _recv(timeout_s: float = 30.0) -> dict:
|
||||
"""Read JSON-RPC responses, skipping notification lines (no 'id' field)."""
|
||||
while True:
|
||||
line = await asyncio.wait_for(proc.stdout.readline(), timeout=timeout_s)
|
||||
if not line:
|
||||
# stdout EOF = child exited. Wait briefly for the stderr
|
||||
# drain to catch up so we capture the real failure reason
|
||||
# (which often arrives a few ms after stdout closes).
|
||||
try:
|
||||
await asyncio.wait_for(asyncio.shield(stderr_task), timeout=1.0)
|
||||
except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
|
||||
pass
|
||||
tail = "".join(stderr_tail[-10:]).strip()
|
||||
raise HTTPException(
|
||||
status_code=502,
|
||||
detail=f"MCP stdio process exited unexpectedly{': ' + tail if tail else ''}",
|
||||
)
|
||||
stripped = line.decode(errors="replace").strip()
|
||||
if not stripped:
|
||||
continue
|
||||
try:
|
||||
data = json.loads(stripped)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
if "id" in data:
|
||||
return data
|
||||
|
||||
try:
|
||||
await _send({
|
||||
"jsonrpc": "2.0", "id": 1, "method": "initialize",
|
||||
"params": {
|
||||
"protocolVersion": "2025-03-26",
|
||||
"capabilities": {},
|
||||
"clientInfo": {"name": "self-swarm", "version": "0.1.0"},
|
||||
},
|
||||
})
|
||||
# First response is the slow one. On Windows with a cold npx cache,
|
||||
# `npx -y <pkg>` has to download the package + transitive deps and
|
||||
# AV-scan every file npm writes; total install time often exceeds
|
||||
# 60 s and occasionally pushes past 90 s. Subsequent reads run
|
||||
# against an already-running server and stay at the default 30 s.
|
||||
await _recv(timeout_s=120.0)
|
||||
|
||||
await _send({"jsonrpc": "2.0", "method": "notifications/initialized"})
|
||||
|
||||
await _send({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}})
|
||||
data = await _recv()
|
||||
|
||||
tools_list = data.get("result", {}).get("tools", [])
|
||||
return [{"name": t.get("name", ""), "description": t.get("description", ""), "inputSchema": t.get("inputSchema")} for t in tools_list]
|
||||
|
||||
except HTTPException as e:
|
||||
# Heal-on-corrupt-npx-cache still triggers from the EOF branch,
|
||||
# which now includes the full stderr tail in `e.detail`; so the
|
||||
# ERR_MODULE_NOT_FOUND signature is still discoverable here.
|
||||
if _attempt == 0 and _try_heal_npx_cache(str(e.detail) if e.detail is not None else ""):
|
||||
return await _discover_mcp_tools_stdio(command, args, env, _attempt=1)
|
||||
raise
|
||||
except asyncio.TimeoutError:
|
||||
# Most common cause: cold npx cache on Windows. The npm install
|
||||
# persists across attempts, so a retry usually finishes against a
|
||||
# warm cache. Surface npx's own progress line if we have one; it
|
||||
# makes the cause obvious ("downloading X...") instead of opaque.
|
||||
tail_text = "".join(stderr_tail[-5:]).strip()
|
||||
detail = "MCP discovery timed out; the server may still be downloading on first run"
|
||||
if tail_text:
|
||||
preview = tail_text[-200:].replace("\n", " ").strip()
|
||||
detail += f" (last output: {preview})"
|
||||
detail += ". Try again in a moment."
|
||||
raise HTTPException(status_code=504, detail=detail)
|
||||
finally:
|
||||
stderr_task.cancel()
|
||||
try:
|
||||
await stderr_task
|
||||
except (asyncio.CancelledError, Exception):
|
||||
pass
|
||||
try:
|
||||
proc.stdin.close()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
proc.terminate()
|
||||
await asyncio.wait_for(proc.wait(), timeout=5.0)
|
||||
except Exception:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
@tools_lib.router.post("/{tool_id}/discover")
|
||||
async def discover_tools(tool_id: str):
|
||||
tool = _load(tool_id)
|
||||
@@ -953,40 +356,6 @@ async def discover_tools(tool_id: str):
|
||||
_m365_login_processes: dict[str, dict] = {} # tool_id -> {proc, device_code, status, email}
|
||||
|
||||
|
||||
def _m365_server_script() -> str:
|
||||
"""Return the on-disk path to the bundled MS365 MCP server entry.
|
||||
|
||||
v1.0.26 replaced the heavy backend/npm-servers/softeria-ms-365-mcp-server/
|
||||
node_modules tree (~93MB / 11k files) with a single esbuild bundle at
|
||||
backend/mcp-bundles/softeria-ms-365-mcp-server/dist/index.js (4.7MB).
|
||||
The new path mirrors the SDK's internal layout (dist/index.js + sibling
|
||||
package.json) because cli.js reads __dirname/../package.json for the
|
||||
--version flag; see scripts/build-app.sh `build_mcp_bundle_dir`.
|
||||
"""
|
||||
_backend = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
bundle = os.path.join(
|
||||
_backend, "mcp-bundles", "softeria-ms-365-mcp-server", "dist", "index.js",
|
||||
)
|
||||
if os.path.isfile(bundle):
|
||||
return bundle
|
||||
# Fallback for any user still on a v1.0.25 install whose backend/ folder
|
||||
# was left over from before the bundle migration. Will return the legacy
|
||||
# path; if that doesn't exist either, the caller raises a clear error.
|
||||
return os.path.join(
|
||||
_backend, "npm-servers", "softeria-ms-365-mcp-server",
|
||||
"node_modules", "@softeria", "ms-365-mcp-server", "dist", "index.js",
|
||||
)
|
||||
|
||||
|
||||
def _m365_cache_env() -> dict[str, str]:
|
||||
cache_dir = os.path.join(os.path.expanduser("~"), ".openswarm")
|
||||
os.makedirs(cache_dir, exist_ok=True)
|
||||
return {
|
||||
"MS365_MCP_TOKEN_CACHE_PATH": os.path.join(cache_dir, "ms365-token-cache.json"),
|
||||
"MS365_MCP_SELECTED_ACCOUNT_PATH": os.path.join(cache_dir, "ms365-selected-account.json"),
|
||||
}
|
||||
|
||||
|
||||
@tools_lib.router.post("/{tool_id}/m365/device-login")
|
||||
async def m365_device_login(tool_id: str):
|
||||
"""Start a Microsoft 365 device-code login.
|
||||
@@ -1158,26 +527,6 @@ async def oauth_disconnect(tool_id: str):
|
||||
return {"ok": True, "tool": tool.model_dump()}
|
||||
|
||||
|
||||
# Tool name → provider key for the OAuth helper service. All providers go
|
||||
# through the Fly cloud-proxy so client_secret values never ship inside the
|
||||
# desktop binary. v1.0.28 was the last release that used a local Google
|
||||
# callback with the client_secret in backend/.env.
|
||||
_TOOL_NAME_TO_PROVIDER = {
|
||||
"airtable": "airtable",
|
||||
"hubspot": "hubspot",
|
||||
"discord": "discord",
|
||||
"notion": "notion",
|
||||
# Built-in Google tool's name is "Google Workspace"; accept the bare
|
||||
# "google" alias too for forward compatibility.
|
||||
"google workspace": "google",
|
||||
"google": "google",
|
||||
}
|
||||
|
||||
|
||||
def _proxied_provider_for(tool: ToolDefinition) -> Optional[str]:
|
||||
return _TOOL_NAME_TO_PROVIDER.get(tool.name.lower())
|
||||
|
||||
|
||||
@tools_lib.router.post("/{tool_id}/oauth/start")
|
||||
async def oauth_start(tool_id: str):
|
||||
"""Return the OAuth start URL for this tool. All built-in providers
|
||||
@@ -1272,116 +621,6 @@ async def oauth_cloud_claim(
|
||||
return _connected_html()
|
||||
|
||||
|
||||
def _persist_cloud_tokens(tool: ToolDefinition, tokens: dict) -> None:
|
||||
"""Normalise the cloud's claim response into tool.oauth_tokens.
|
||||
|
||||
Per-provider shaping mirrors what the v1.0.25 local-callback flow used
|
||||
to write; the rest of the app (refresh helpers, MCP env injection)
|
||||
expects exactly this shape.
|
||||
"""
|
||||
name = tool.name.lower()
|
||||
if name == "discord":
|
||||
new_guilds = (tokens.get("_guilds") or []) if isinstance(tokens, dict) else []
|
||||
existing = tool.oauth_tokens.get("guilds") or []
|
||||
for g in new_guilds:
|
||||
if g.get("id") and not any(e.get("id") == g["id"] for e in existing):
|
||||
existing.append({"id": g["id"], "name": g.get("name", "")})
|
||||
tool.oauth_tokens = {"guilds": existing}
|
||||
names = ", ".join(g.get("name", "") for g in existing if g.get("name"))
|
||||
tool.connected_account_email = (
|
||||
f"{len(existing)} server{'s' if len(existing) != 1 else ''}"
|
||||
+ (f" · {names}" if names else "")
|
||||
)
|
||||
elif name == "notion":
|
||||
tool.oauth_tokens = {"access_token": tokens.get("access_token", "")}
|
||||
tool.connected_account_email = tokens.get("workspace_name", "Notion workspace")
|
||||
else:
|
||||
tool.oauth_tokens = {
|
||||
"access_token": tokens.get("access_token", ""),
|
||||
"refresh_token": tokens.get("refresh_token", ""),
|
||||
"token_expiry": time.time() + (tokens.get("expires_in") or 3600),
|
||||
}
|
||||
tool.connected_account_email = (
|
||||
tokens.get("email") # Google (post-userinfo enrichment)
|
||||
or tokens.get("hub_domain") # HubSpot
|
||||
or tokens.get("workspace_name")
|
||||
or f"{tool.name} account"
|
||||
)
|
||||
tool.auth_type = "oauth2"
|
||||
tool.auth_status = "connected"
|
||||
|
||||
|
||||
async def _refresh_via_proxy(provider: str, tool: ToolDefinition, default_expiry: int) -> Optional[str]:
|
||||
"""Refresh an OAuth access_token by POSTing the refresh_token to the
|
||||
helper service. Per-provider wrappers below pass a default expires_in
|
||||
fallback for providers that don't return one.
|
||||
"""
|
||||
if tool.auth_type != "oauth2":
|
||||
return None
|
||||
refresh_token = tool.oauth_tokens.get("refresh_token")
|
||||
if not refresh_token:
|
||||
return None
|
||||
expiry = tool.oauth_tokens.get("token_expiry", 0)
|
||||
if time.time() < expiry - 60:
|
||||
return tool.oauth_tokens.get("access_token")
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||||
resp = await client.post(
|
||||
f"{OPENSWARM_OAUTH_BASE_URL}/api/oauth/{provider}/refresh",
|
||||
json={"refresh_token": refresh_token},
|
||||
)
|
||||
if resp.status_code == 401:
|
||||
# Provider rejected; user revoked at the provider's side. Mark
|
||||
# as needing re-auth so the UI prompts a Reconnect.
|
||||
tool.auth_status = "expired"
|
||||
_save(tool)
|
||||
logger.warning(f"{provider} refresh rejected (user revoked); marking tool as expired")
|
||||
return None
|
||||
if resp.status_code != 200:
|
||||
logger.warning(f"{provider} cloud refresh failed: HTTP %d %s", resp.status_code, resp.text[:200])
|
||||
return None
|
||||
|
||||
data = (resp.json() or {}).get("tokens") or {}
|
||||
new_token = data.get("access_token", "")
|
||||
if not new_token:
|
||||
return None
|
||||
tool.oauth_tokens["access_token"] = new_token
|
||||
tool.oauth_tokens["token_expiry"] = time.time() + (data.get("expires_in") or default_expiry)
|
||||
if data.get("refresh_token"):
|
||||
# Some providers (HubSpot, Airtable) rotate refresh_tokens on every
|
||||
# refresh. Persist the new one or future refreshes will fail.
|
||||
tool.oauth_tokens["refresh_token"] = data["refresh_token"]
|
||||
# Backfill identity label on first successful refresh after upgrade.
|
||||
if not tool.connected_account_email and data.get("email"):
|
||||
tool.connected_account_email = data["email"]
|
||||
_save(tool)
|
||||
return new_token
|
||||
except Exception as e:
|
||||
logger.warning(f"{provider} cloud refresh exception for tool {tool.id}: {e}")
|
||||
return None
|
||||
|
||||
|
||||
async def refresh_google_token(tool: ToolDefinition) -> Optional[str]:
|
||||
"""Refresh an expired Google access_token via the Fly cloud-proxy.
|
||||
|
||||
The client_secret never leaves Fly; desktop only POSTs the
|
||||
refresh_token. Same pattern as Airtable/HubSpot. Pre-v1.0.29 builds
|
||||
held the secret in their bundled .env; v1.0.29 removed it.
|
||||
"""
|
||||
return await _refresh_via_proxy("google", tool, default_expiry=3600)
|
||||
|
||||
|
||||
async def refresh_airtable_token(tool: ToolDefinition) -> Optional[str]:
|
||||
"""Refresh an expired Airtable OAuth access_token."""
|
||||
return await _refresh_via_proxy("airtable", tool, default_expiry=7200)
|
||||
|
||||
|
||||
async def refresh_hubspot_token(tool: ToolDefinition) -> Optional[str]:
|
||||
"""Refresh an expired HubSpot OAuth access_token."""
|
||||
return await _refresh_via_proxy("hubspot", tool, default_expiry=1800)
|
||||
|
||||
|
||||
@tools_lib.router.post("/google-oauth-token")
|
||||
async def google_oauth_token_proxy(request: Request):
|
||||
"""Local mimic of Google's OAuth2 token endpoint for the
|
||||
|
||||
Reference in New Issue
Block a user