From 8c7cda89dc3a5848fe54872d17a661ecf4d9d4a8 Mon Sep 17 00:00:00 2001 From: zhayujie Date: Mon, 13 Jul 2026 12:00:14 +0800 Subject: [PATCH] feat(mcp): support OAuth authorization for remote MCP servers --- agent/tools/mcp/mcp_client.py | 165 +++++++++++- agent/tools/mcp/mcp_oauth.py | 466 ++++++++++++++++++++++++++++++++++ agent/tools/tool_manager.py | 41 ++- channel/web/web_channel.py | 59 +++++ config.py | 1 + docs/ja/tools/mcp.mdx | 22 ++ docs/tools/mcp.mdx | 22 ++ docs/zh/tools/mcp.mdx | 22 ++ 8 files changed, 792 insertions(+), 6 deletions(-) create mode 100644 agent/tools/mcp/mcp_oauth.py diff --git a/agent/tools/mcp/mcp_client.py b/agent/tools/mcp/mcp_client.py index 81d81004..e67cbbdc 100644 --- a/agent/tools/mcp/mcp_client.py +++ b/agent/tools/mcp/mcp_client.py @@ -21,6 +21,48 @@ from common.log import logger _STREAMABLE_HTTP_ALIASES = {"streamable-http", "streamable_http", "streamablehttp", "http"} +# Optional callback invoked after an OAuth authorization completes, so the +# tool manager can bring the newly-authorized server online. Signature: +# reload_fn(server_name: str) -> None. Installed by the tool manager. +_reload_callback = None + + +def set_reload_callback(fn) -> None: + """Register a callback fired after a server's OAuth flow succeeds.""" + global _reload_callback + _reload_callback = fn + + +def notify_server_authorized(server_name: str) -> None: + """Called by the web callback once tokens are stored for a server.""" + fn = _reload_callback + if fn is None: + logger.debug(f"[MCP:{server_name}] Authorized but no reload callback registered") + return + try: + fn(server_name) + except Exception as e: + logger.warning(f"[MCP:{server_name}] reload callback failed: {e}") + + +def _oauth_redirect_uri() -> str: + """Build the OAuth redirect URI served by the web console callback. + + Priority: explicit mcp_oauth_redirect_base config, otherwise the local + web console address (127.0.0.1:). Both point at the shared + /mcp/oauth/callback route. + """ + try: + from config import conf + base = (conf().get("mcp_oauth_redirect_base") or "").strip().rstrip("/") + if not base: + port = int(os.environ.get("COW_WEB_PORT") or conf().get("web_port", 9899)) + base = f"http://127.0.0.1:{port}" + except Exception: + base = "http://127.0.0.1:9899" + return f"{base}/mcp/oauth/callback" + + class McpClient: """Single MCP Server client supporting stdio, SSE and Streamable HTTP transports.""" @@ -56,6 +98,13 @@ class McpClient: self._http_headers: dict = {} # extra headers from user config (e.g. Authorization) self._http_session_id: Optional[str] = None # Mcp-Session-Id assigned by the server + # OAuth state (streamable-http only). Lazily created when the server + # responds with 401 and the user has not supplied a static token. + self._oauth = None # OAuthHandler instance + # Set to True once a 401 could not be satisfied and the user must + # complete the browser authorization. Callers can surface this state. + self.needs_auth: bool = False + # Shared state self._next_id = 1 self._id_lock = threading.Lock() @@ -325,13 +374,118 @@ class McpClient: if isinstance(extra_headers, dict): self._http_headers = {str(k): str(v) for k, v in extra_headers.items()} + # Restore any previously stored OAuth credentials for this server so a + # restart reuses the token instead of forcing re-authorization. + self._maybe_load_oauth() + return self._handshake() + # ------------------------------------------------------------------ + # OAuth helpers (streamable-http only) + # ------------------------------------------------------------------ + + def _has_static_auth(self) -> bool: + """True when the user supplied their own Authorization header.""" + return any(k.lower() == "authorization" for k in self._http_headers) + + def _maybe_load_oauth(self) -> None: + """Attach an OAuthHandler when stored credentials exist for this server.""" + if self._has_static_auth(): + return + try: + from agent.tools.mcp.mcp_oauth import OAuthHandler, load_server_record + except Exception: + return + rec = load_server_record(self.name) + # Only create a handler when we have something to reuse; otherwise it + # is created lazily on the first 401. + if rec.get("access_token") or rec.get("client_id"): + self._oauth = OAuthHandler( + server_name=self.name, + resource_url=self._http_url, + redirect_uri=_oauth_redirect_uri(), + scope=self.config.get("scope", ""), + ) + + def _current_bearer(self) -> Optional[str]: + """Return a valid access token, refreshing if needed.""" + if self._oauth is None: + return None + return self._oauth.get_valid_access_token() + + def _begin_oauth(self, www_authenticate: str = "") -> None: + """Kick off the OAuth flow after a 401: discover, register, prompt user.""" + if self._has_static_auth(): + return + try: + from agent.tools.mcp.mcp_oauth import OAuthHandler + except Exception as e: + logger.warning(f"[MCP:{self.name}] OAuth module unavailable: {e}") + return + + if self._oauth is None: + self._oauth = OAuthHandler( + server_name=self.name, + resource_url=self._http_url, + redirect_uri=_oauth_redirect_uri(), + scope=self.config.get("scope", ""), + ) + + if not self._oauth.ensure_registered(www_authenticate): + logger.warning( + f"[MCP:{self.name}] OAuth discovery/registration failed; " + f"cannot authorize automatically" + ) + return + + auth_url = self._oauth.build_authorization_url() + if not auth_url: + logger.warning(f"[MCP:{self.name}] Failed to build authorization URL") + return + + self.needs_auth = True + logger.warning( + f"[MCP:{self.name}] ⚠️ Authorization required. Open this URL in a " + f"browser to authorize, then this server will come online automatically:\n" + f" {auth_url}" + ) + # On a machine with a local browser (desktop/dev), open it directly. + if os.environ.get("COW_DESKTOP") == "1" or not os.environ.get("COW_HEADLESS"): + try: + import webbrowser + webbrowser.open(auth_url) + except Exception: + pass + def _streamable_http_send(self, message: dict) -> dict: """POST a JSON-RPC request and return the response (JSON or SSE-wrapped).""" return self._streamable_http_post(message, expect_response=True) - def _streamable_http_post(self, message: dict, expect_response: bool) -> dict: + def _handle_401(self, err, message: dict, expect_response: bool, retried: bool) -> dict: + """Handle a 401: refresh the token and retry once, else begin OAuth.""" + www_auth = "" + try: + www_auth = err.headers.get("WWW-Authenticate", "") or "" + except Exception: + pass + try: + err.read() + except Exception: + pass + + # First try a silent refresh with the stored refresh token. + if not retried and self._oauth is not None and self._oauth.refresh(): + logger.info(f"[MCP:{self.name}] Token refreshed after 401, retrying") + return self._streamable_http_post(message, expect_response, _retried=True) + + # No usable token — start (or restart) the interactive OAuth flow. + self._begin_oauth(www_auth) + raise IOError( + f"[MCP:{self.name}] streamable-http HTTP 401: authorization required " + f"(complete the OAuth flow to enable this server)" + ) + + def _streamable_http_post(self, message: dict, expect_response: bool, _retried: bool = False) -> dict: """ POST a JSON-RPC message over Streamable HTTP. @@ -351,6 +505,12 @@ class McpClient: if sid: headers["Mcp-Session-Id"] = sid headers.update(self._http_headers) + # Inject OAuth bearer token when we have one (unless the user set a + # static Authorization header, which takes precedence). + if not self._has_static_auth(): + token = self._current_bearer() + if token: + headers["Authorization"] = f"Bearer {token}" req = urllib.request.Request( self._http_url, @@ -362,6 +522,9 @@ class McpClient: try: resp = urllib.request.urlopen(req, timeout=30) except urllib.error.HTTPError as e: + # 401 is the spec-compliant "needs authorization" signal. + if e.code == 401 and not self._has_static_auth(): + return self._handle_401(e, message, expect_response, _retried) # Surface the server-provided error body for easier debugging detail = "" try: diff --git a/agent/tools/mcp/mcp_oauth.py b/agent/tools/mcp/mcp_oauth.py new file mode 100644 index 00000000..90b79a16 --- /dev/null +++ b/agent/tools/mcp/mcp_oauth.py @@ -0,0 +1,466 @@ +""" +MCP OAuth 2.1 client (authorization code + PKCE) with zero external deps. + +Implements the subset of the MCP authorization spec needed to connect to +remote MCP servers that guard their endpoint behind OAuth (e.g. Xmind): + + 1. Metadata discovery via RFC 9728 (protected-resource) + RFC 8414 + (authorization-server) .well-known documents. + 2. Dynamic Client Registration (RFC 7591) to obtain a client_id. + 3. PKCE (RFC 7636, S256) authorization-code flow. + 4. Token exchange + refresh, persisted to ~/.cow/mcp_oauth.json. + +The actual browser round-trip is completed out-of-band: McpClient generates +an authorization URL, the user opens it, and the web console callback +(/mcp/oauth/callback) feeds the returned code back into finish_authorization(). +""" + +import base64 +import hashlib +import json +import os +import secrets +import threading +import time +import urllib.parse +import urllib.request +import urllib.error +from typing import Optional + +from common.log import logger + + +# ------------------------------------------------------------------ +# Token store: ~/.cow/mcp_oauth.json {server_name: {...credentials...}} +# ------------------------------------------------------------------ + +_STORE_LOCK = threading.Lock() + + +def _store_path() -> str: + base = os.path.expanduser("~/.cow") + try: + os.makedirs(base, exist_ok=True) + except OSError: + pass + return os.path.join(base, "mcp_oauth.json") + + +def _load_store() -> dict: + path = _store_path() + if not os.path.exists(path): + return {} + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + return data if isinstance(data, dict) else {} + except Exception as e: + logger.warning(f"[MCP-OAuth] Failed to read token store: {e}") + return {} + + +def _save_store(store: dict) -> None: + path = _store_path() + tmp = f"{path}.tmp" + try: + with open(tmp, "w", encoding="utf-8") as f: + json.dump(store, f, ensure_ascii=False, indent=2) + os.replace(tmp, path) + # Credentials file: restrict to owner read/write when possible. + try: + os.chmod(path, 0o600) + except OSError: + pass + except Exception as e: + logger.warning(f"[MCP-OAuth] Failed to persist token store: {e}") + + +def load_server_record(server_name: str) -> dict: + with _STORE_LOCK: + return dict(_load_store().get(server_name, {})) + + +def save_server_record(server_name: str, record: dict) -> None: + with _STORE_LOCK: + store = _load_store() + store[server_name] = record + _save_store(store) + + +def clear_server_record(server_name: str) -> None: + with _STORE_LOCK: + store = _load_store() + if server_name in store: + store.pop(server_name, None) + _save_store(store) + + +# ------------------------------------------------------------------ +# Pending authorizations, keyed by the OAuth `state` param. +# Populated when an authorization URL is generated; consumed by the +# web callback when the browser redirects back with ?code&state. +# ------------------------------------------------------------------ + +_PENDING_LOCK = threading.Lock() +_PENDING: dict = {} # state -> {"handler": OAuthHandler, "created": ts} +_PENDING_TTL = 600 # seconds + + +def _register_pending(state: str, handler: "OAuthHandler") -> None: + with _PENDING_LOCK: + _prune_pending_locked() + _PENDING[state] = {"handler": handler, "created": time.time()} + + +def _prune_pending_locked() -> None: + now = time.time() + stale = [s for s, v in _PENDING.items() if now - v["created"] > _PENDING_TTL] + for s in stale: + _PENDING.pop(s, None) + + +def pop_pending(state: str) -> Optional["OAuthHandler"]: + with _PENDING_LOCK: + _prune_pending_locked() + entry = _PENDING.pop(state, None) + return entry["handler"] if entry else None + + +def has_pending() -> bool: + with _PENDING_LOCK: + _prune_pending_locked() + return bool(_PENDING) + + +# ------------------------------------------------------------------ +# HTTP helpers (stdlib only) +# ------------------------------------------------------------------ + +_UA = "CowAgent-MCP-OAuth/1.0" + + +def _http_get_json(url: str, timeout: int = 15) -> Optional[dict]: + req = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": _UA}) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw = resp.read().decode("utf-8") + return json.loads(raw) + except urllib.error.HTTPError as e: + logger.debug(f"[MCP-OAuth] GET {url} -> HTTP {e.code}") + return None + except Exception as e: + logger.debug(f"[MCP-OAuth] GET {url} failed: {e}") + return None + + +def _http_post_form(url: str, fields: dict, timeout: int = 20) -> dict: + body = urllib.parse.urlencode(fields).encode("utf-8") + req = urllib.request.Request( + url, + data=body, + method="POST", + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + "User-Agent": _UA, + }, + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw = resp.read().decode("utf-8") + return json.loads(raw) if raw else {} + + +def _http_post_json(url: str, payload: dict, timeout: int = 20) -> dict: + body = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + url, + data=body, + method="POST", + headers={ + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": _UA, + }, + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw = resp.read().decode("utf-8") + return json.loads(raw) if raw else {} + + +# ------------------------------------------------------------------ +# Discovery (RFC 9728 + RFC 8414) +# ------------------------------------------------------------------ + +def _origin(url: str) -> str: + p = urllib.parse.urlparse(url) + return f"{p.scheme}://{p.netloc}" + + +def discover_metadata(resource_url: str, www_authenticate: str = "") -> Optional[dict]: + """ + Resolve the authorization server metadata for a protected MCP resource. + + Returns a dict with at least authorization_endpoint + token_endpoint, + plus registration_endpoint when the server supports DCR. Returns None + when discovery fails. + """ + as_metadata_url = _parse_resource_metadata_url(www_authenticate) + + # 1) Protected-resource metadata (RFC 9728) to locate the auth server. + auth_server = None + prm = None + if as_metadata_url: + prm = _http_get_json(as_metadata_url) + if prm is None: + origin = _origin(resource_url) + prm = _http_get_json(f"{origin}/.well-known/oauth-protected-resource") + if prm and isinstance(prm.get("authorization_servers"), list) and prm["authorization_servers"]: + auth_server = prm["authorization_servers"][0] + + # 2) Authorization-server metadata (RFC 8414). Fall back to the resource + # origin when the resource did not advertise a separate auth server. + base = auth_server or _origin(resource_url) + asm = _fetch_as_metadata(base) + if not asm: + return None + + if not asm.get("authorization_endpoint") or not asm.get("token_endpoint"): + logger.warning("[MCP-OAuth] Authorization server metadata missing required endpoints") + return None + + # Derive the scope to request. Prefer the resource's required_scopes + # (RFC 9728), then its scopes_supported, then the auth server's + # scopes_supported. Stored so callers don't have to configure it. + discovered_scope = "" + if prm: + scopes = prm.get("required_scopes") or prm.get("scopes_supported") + if isinstance(scopes, list) and scopes: + discovered_scope = " ".join(str(s) for s in scopes) + if not discovered_scope and isinstance(asm.get("scopes_supported"), list) and asm["scopes_supported"]: + discovered_scope = " ".join(str(s) for s in asm["scopes_supported"]) + if discovered_scope: + asm["_discovered_scope"] = discovered_scope + return asm + + +def _parse_resource_metadata_url(www_authenticate: str) -> Optional[str]: + """Extract resource_metadata="..." from a WWW-Authenticate: Bearer header.""" + if not www_authenticate: + return None + # naive but sufficient parse for `resource_metadata="URL"` + marker = "resource_metadata=" + idx = www_authenticate.find(marker) + if idx < 0: + return None + rest = www_authenticate[idx + len(marker):].strip() + if rest.startswith('"'): + end = rest.find('"', 1) + return rest[1:end] if end > 0 else None + # unquoted, up to comma/space + for sep in (",", " "): + if sep in rest: + rest = rest.split(sep, 1)[0] + return rest or None + + +def _fetch_as_metadata(base: str) -> Optional[dict]: + """Try both RFC 8414 and OIDC well-known locations.""" + base = base.rstrip("/") + candidates = [ + f"{base}/.well-known/oauth-authorization-server", + f"{base}/.well-known/openid-configuration", + ] + for url in candidates: + data = _http_get_json(url) + if data and data.get("authorization_endpoint"): + return data + return None + + +# ------------------------------------------------------------------ +# PKCE +# ------------------------------------------------------------------ + +def _b64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=") + + +def _make_pkce() -> tuple: + verifier = _b64url(secrets.token_bytes(32)) + challenge = _b64url(hashlib.sha256(verifier.encode("ascii")).digest()) + return verifier, challenge + + +# ------------------------------------------------------------------ +# OAuthHandler: per-server OAuth state machine +# ------------------------------------------------------------------ + +class OAuthHandler: + """Drives the OAuth flow and token lifecycle for a single MCP server.""" + + def __init__(self, server_name: str, resource_url: str, redirect_uri: str, + scope: str = "", client_name: str = "CowAgent"): + self.server_name = server_name + self.resource_url = resource_url + self.redirect_uri = redirect_uri + self.scope = scope + self.client_name = client_name + + rec = load_server_record(server_name) + self.metadata: dict = rec.get("metadata", {}) + self.client_id: Optional[str] = rec.get("client_id") + self.client_secret: Optional[str] = rec.get("client_secret") + self.access_token: Optional[str] = rec.get("access_token") + self.refresh_token: Optional[str] = rec.get("refresh_token") + self.expires_at: float = float(rec.get("expires_at", 0) or 0) + self._verifier: Optional[str] = None + + # --- persistence ------------------------------------------------- + + def _persist(self) -> None: + save_server_record(self.server_name, { + "resource_url": self.resource_url, + "metadata": self.metadata, + "client_id": self.client_id, + "client_secret": self.client_secret, + "access_token": self.access_token, + "refresh_token": self.refresh_token, + "expires_at": self.expires_at, + }) + + # --- token access ------------------------------------------------ + + def get_valid_access_token(self, leeway: int = 60) -> Optional[str]: + """Return a usable access token, refreshing proactively when near expiry.""" + if not self.access_token: + return None + if self.expires_at and time.time() >= self.expires_at - leeway: + if not self.refresh(): + return None + return self.access_token + + def refresh(self) -> bool: + """Refresh the access token using the stored refresh token.""" + if not self.refresh_token or not self.metadata.get("token_endpoint"): + return False + fields = { + "grant_type": "refresh_token", + "refresh_token": self.refresh_token, + "client_id": self.client_id or "", + } + if self.client_secret: + fields["client_secret"] = self.client_secret + try: + resp = _http_post_form(self.metadata["token_endpoint"], fields) + except Exception as e: + logger.warning(f"[MCP-OAuth:{self.server_name}] refresh failed: {e}") + return False + return self._absorb_token_response(resp) + + # --- authorization-code flow ------------------------------------ + + def ensure_registered(self, www_authenticate: str = "") -> bool: + """Discover metadata + register a client if not already done.""" + if not self.metadata.get("authorization_endpoint"): + meta = discover_metadata(self.resource_url, www_authenticate) + if not meta: + return False + self.metadata = meta + # Adopt the scope discovered from metadata when the user didn't set one. + if not self.scope and self.metadata.get("_discovered_scope"): + self.scope = self.metadata["_discovered_scope"] + logger.info(f"[MCP-OAuth:{self.server_name}] Using discovered scope: {self.scope}") + if not self.client_id: + if not self._register_client(): + return False + self._persist() + return True + + def _register_client(self) -> bool: + reg_endpoint = self.metadata.get("registration_endpoint") + if not reg_endpoint: + logger.warning( + f"[MCP-OAuth:{self.server_name}] No registration_endpoint; " + f"DCR unavailable. Provide client_id manually." + ) + return False + payload = { + "client_name": self.client_name, + "redirect_uris": [self.redirect_uri], + "grant_types": ["authorization_code", "refresh_token"], + "response_types": ["code"], + "token_endpoint_auth_method": "none", + } + if self.scope: + payload["scope"] = self.scope + try: + resp = _http_post_json(reg_endpoint, payload) + except Exception as e: + logger.warning(f"[MCP-OAuth:{self.server_name}] DCR failed: {e}") + return False + client_id = resp.get("client_id") + if not client_id: + logger.warning(f"[MCP-OAuth:{self.server_name}] DCR returned no client_id") + return False + self.client_id = client_id + self.client_secret = resp.get("client_secret") + logger.info(f"[MCP-OAuth:{self.server_name}] Registered client_id={client_id}") + return True + + def build_authorization_url(self) -> Optional[str]: + """Create an authorization URL and register this handler as pending.""" + if not self.metadata.get("authorization_endpoint") or not self.client_id: + return None + self._verifier, challenge = _make_pkce() + state = secrets.token_urlsafe(24) + params = { + "response_type": "code", + "client_id": self.client_id, + "redirect_uri": self.redirect_uri, + "code_challenge": challenge, + "code_challenge_method": "S256", + "state": state, + } + if self.scope: + params["scope"] = self.scope + # Advertise the resource we intend to access (RFC 8707). + params["resource"] = self.resource_url + _register_pending(state, self) + return f"{self.metadata['authorization_endpoint']}?{urllib.parse.urlencode(params)}" + + def finish_authorization(self, code: str) -> bool: + """Exchange an authorization code for tokens.""" + if not self.metadata.get("token_endpoint") or not self._verifier: + return False + fields = { + "grant_type": "authorization_code", + "code": code, + "redirect_uri": self.redirect_uri, + "client_id": self.client_id or "", + "code_verifier": self._verifier, + "resource": self.resource_url, + } + if self.client_secret: + fields["client_secret"] = self.client_secret + try: + resp = _http_post_form(self.metadata["token_endpoint"], fields) + except Exception as e: + logger.warning(f"[MCP-OAuth:{self.server_name}] token exchange failed: {e}") + return False + ok = self._absorb_token_response(resp) + self._verifier = None + return ok + + def _absorb_token_response(self, resp: dict) -> bool: + access = resp.get("access_token") + if not access: + logger.warning(f"[MCP-OAuth:{self.server_name}] token response missing access_token: {resp}") + return False + self.access_token = access + if resp.get("refresh_token"): + self.refresh_token = resp["refresh_token"] + expires_in = resp.get("expires_in") + self.expires_at = time.time() + int(expires_in) if expires_in else 0 + self._persist() + logger.info(f"[MCP-OAuth:{self.server_name}] Access token stored") + return True diff --git a/agent/tools/tool_manager.py b/agent/tools/tool_manager.py index 9f9db462..e282a15e 100644 --- a/agent/tools/tool_manager.py +++ b/agent/tools/tool_manager.py @@ -466,21 +466,30 @@ class ToolManager: the others, and never raises out of the worker thread. """ try: - from agent.tools.mcp.mcp_client import McpClient, McpClientRegistry + from agent.tools.mcp.mcp_client import McpClient, McpClientRegistry, set_reload_callback from agent.tools.mcp.mcp_tool import McpTool registry = McpClientRegistry() self._mcp_registry = registry + # Let the OAuth web callback bring a server online once authorized. + set_reload_callback(self.reload_mcp_server) for cfg in mcp_servers_config: server_name = cfg.get("name", "") try: client = McpClient(cfg) if not client.initialize(): - self._mcp_status[server_name] = "failed" - logger.warning( - f"[MCP] Server '{server_name}' failed to initialize — skipping" - ) + if getattr(client, "needs_auth", False): + self._mcp_status[server_name] = "needs_auth" + logger.info( + f"[MCP] Server '{server_name}' needs authorization — " + f"waiting for the user to complete the OAuth flow" + ) + else: + self._mcp_status[server_name] = "failed" + logger.warning( + f"[MCP] Server '{server_name}' failed to initialize — skipping" + ) continue tool_schemas = client.list_tools() @@ -518,6 +527,28 @@ class ToolManager: except Exception as e: logger.warning(f"[ToolManager] MCP background loader crashed: {e}") + def reload_mcp_server(self, server_name: str) -> None: + """Re-initialize a single MCP server (e.g. after OAuth authorization). + + Tears down any existing client for the server and starts it again in + the background, so a freshly-stored access token is picked up and the + server's tools become available on the next message. + """ + with self._mcp_lock: + cfg = self._mcp_active_configs.get(server_name) + if not cfg: + logger.warning(f"[MCP] reload requested for unknown server '{server_name}'") + return + logger.info(f"[MCP] Reloading server '{server_name}' after authorization") + self._teardown_mcp_server(server_name) + self._mcp_status[server_name] = "pending" + threading.Thread( + target=self._load_mcp_tools_async, + args=([cfg],), + daemon=True, + name=f"mcp-reload-{server_name}", + ).start() + def list_mcp_status(self) -> dict: """Return {server_name: status} snapshot for UI / debugging.""" return dict(self._mcp_status) diff --git a/channel/web/web_channel.py b/channel/web/web_channel.py index 481a4108..7f0a6e3d 100644 --- a/channel/web/web_channel.py +++ b/channel/web/web_channel.py @@ -1320,6 +1320,7 @@ class WebChannel(ChatChannel): '/api/messages/delete', 'MessageDeleteHandler', '/api/logs', 'LogsHandler', '/api/version', 'VersionHandler', + '/mcp/oauth/callback', 'McpOAuthCallbackHandler', '/assets/(.*)', 'AssetsHandler', ) app = web.application(urls, globals(), autoreload=False) @@ -1383,6 +1384,64 @@ class HealthHandler: return json.dumps({"status": "ok"}) +class McpOAuthCallbackHandler: + """OAuth redirect target for MCP servers requiring authorization. + + The browser lands here after the user authorizes a remote MCP server. + We exchange the authorization code for tokens and bring the server + online. Unauthenticated by design: the OAuth `state` param is the + single-use secret that binds this request to a pending authorization. + """ + + def GET(self): + web.header('Content-Type', 'text/html; charset=utf-8') + params = web.input(code="", state="", error="", error_description="") + + def _page(title: str, message: str) -> str: + return ( + "" + "" + f"{title}" + "" + f"

{title}

{message}

" + ) + + if params.error: + logger.warning(f"[MCP-OAuth] callback error: {params.error} {params.error_description}") + return _page("授权失败", f"{params.error}: {params.error_description or ''}") + + if not params.code or not params.state: + return _page("参数缺失", "回调缺少 code 或 state 参数。") + + try: + from agent.tools.mcp.mcp_oauth import pop_pending + from agent.tools.mcp.mcp_client import notify_server_authorized + except Exception as e: + logger.warning(f"[MCP-OAuth] callback import failed: {e}") + return _page("内部错误", "OAuth 模块不可用。") + + handler = pop_pending(params.state) + if handler is None: + return _page("会话已过期", "授权请求不存在或已过期，请重新触发授权。") + + try: + ok = handler.finish_authorization(params.code) + except Exception as e: + logger.warning(f"[MCP-OAuth] token exchange crashed: {e}") + ok = False + + if not ok: + return _page("授权失败", "换取令牌失败，请重试。") + + notify_server_authorized(handler.server_name) + logger.info(f"[MCP-OAuth] Server '{handler.server_name}' authorized via web callback") + return _page( + "授权成功", + f"MCP 服务 “{handler.server_name}” 已授权，可以返回聊天继续使用了。", + ) + + class AuthCheckHandler: def GET(self): web.header('Content-Type', 'application/json; charset=utf-8') diff --git a/config.py b/config.py index 6fe2db03..6d02e86f 100644 --- a/config.py +++ b/config.py @@ -253,6 +253,7 @@ available_setting = { "web_password": "", # Web console password; empty means no authentication required "web_session_expire_days": 30, # Auth session expiry in days "web_file_serve_root": "~", # Root dir the /api/file endpoint may serve; "/" allows the whole filesystem + "mcp_oauth_redirect_base": "", # Base URL for MCP OAuth callback (e.g. http://your-ip:9899); empty uses local web console "agent": True, # whether to enable Agent mode "agent_workspace": "~/cow", # agent workspace path, used to store skills, memory, etc. "agent_max_context_tokens": 50000, # max context tokens in Agent mode diff --git a/docs/ja/tools/mcp.mdx b/docs/ja/tools/mcp.mdx index efd4a514..236d2211 100644 --- a/docs/ja/tools/mcp.mdx +++ b/docs/ja/tools/mcp.mdx @@ -37,6 +37,7 @@ MCP コミュニティ標準に完全準拠しており、Claude Desktop / Curso | `url` | SSE / Streamable HTTP | リモートエンドポイントの URL（`command` と二者択一） | | `type` | リモート | リモートトランスポート種別：`sse` または `streamable-http`（既定は `sse`） | | `headers` | 任意 | リモートリクエストの追加 HTTP ヘッダ（`Authorization` など）。Streamable HTTP のみ | +| `scope` | 任意 | OAuth スコープ。OAuth 認可が必要なリモート server のみ使用（任意） | | `disabled` | 任意 | `true` のとき該当サーバーをスキップ。一時的に無効化したいときに便利 | ### 完全な例 @@ -79,6 +80,27 @@ Agent は次のように動作します： 1. 既存の MCP 設定ファイルを読み込み、新しい server エントリをマージ（既存の項目は保持） 2. 増分の MCP Server を自動でリロードし、次のメッセージから対応する Tool が利用可能に +## Web 認可（OAuth） + +一部のリモート MCP は OAuth の Web 認可が必要で、そのまま設定すると `401` が返ります。CowAgent は標準的な OAuth フローを内蔵しているため、**token を手動で入力する必要はなく**、通常どおり設定するだけです。例： + +```json +{ + "mcpServers": { + "xmind": { + "type": "streamable-http", + "url": "https://app.xmind.com/api/mcp" + } + } +} +``` + +server の初回ロードで `401` が返ると、認可が自動的に開始されます。ローカル実行では**自動的にブラウザが開き**、サーバー環境では**認可リンクがログに出力**されるので、ブラウザで開いてください。承認すると server はすぐにオンラインになり、token は期限切れ時に自動更新されるため、再認可は不要です。 + +- **Web サービスが必要**：認可コールバックは Web コンソール（既定ポート `9899`）で受け取るため、Web channel が起動している必要があります。 +- **認証情報の保存**：token は `~/.cow/mcp_oauth.json` に永続化され、再起動後も再利用されます。 +- **コールバック URL**：既定は `http://127.0.0.1:9899/mcp/oauth/callback`。サーバーに配置し認可用ブラウザが別の端末にある場合は、`config.json` に `mcp_oauth_redirect_base`（例：`http://あなたのIP:9899`）を設定してください。 + ## 動作の仕組み - **起動時の非同期ロード**：`mcp.json` に設定された全 server はバックグラウンドで非同期に読み込まれ、メインループをブロックしません。会話はすぐに開始できます diff --git a/docs/tools/mcp.mdx b/docs/tools/mcp.mdx index fc320fe0..6c3e0ece 100644 --- a/docs/tools/mcp.mdx +++ b/docs/tools/mcp.mdx @@ -37,6 +37,7 @@ Fully compatible with the MCP community standard, identical to Claude Desktop / | `url` | SSE / Streamable HTTP | Remote endpoint URL (alternative to `command`) | | `type` | Remote | Remote transport type: `sse` or `streamable-http` (defaults to `sse`) | | `headers` | No | Extra HTTP headers for remote requests (e.g. `Authorization`); Streamable HTTP only | +| `scope` | No | OAuth scope, only for remote servers that require OAuth authorization (optional) | | `disabled` | No | When `true`, this server is skipped — handy for temporary disabling | ### Full Example @@ -79,6 +80,27 @@ The Agent will: 1. Read the existing MCP config and merge the new server entry, preserving existing ones 2. Hot-reload the new MCP server, so the corresponding tools become available on the next message +## Web Authorization (OAuth) + +Some remote MCP servers require OAuth web authorization, and connecting to them directly returns `401`. CowAgent has a built-in standard OAuth flow, so **no manual token is needed** — just configure the server normally, for example: + +```json +{ + "mcpServers": { + "xmind": { + "type": "streamable-http", + "url": "https://app.xmind.com/api/mcp" + } + } +} +``` + +When a server returns `401` on its first load, authorization starts automatically: running locally **opens the browser automatically**, while server deployments **print the authorization link to the log** for you to open in a browser. Once you approve, the server comes online immediately; tokens are refreshed automatically on expiry, so you never have to re-authorize. + +- **Requires the web service**: The authorization callback is received by the web console (default port `9899`), so the Web channel must be running. +- **Credential storage**: Tokens are persisted in `~/.cow/mcp_oauth.json` and reused across restarts. +- **Callback URL**: Defaults to `http://127.0.0.1:9899/mcp/oauth/callback`. If deployed on a server with the authorizing browser on another device, set `mcp_oauth_redirect_base` in `config.json` (e.g. `http://YOUR_IP:9899`). + ## How It Works - **Async loading at startup**: All servers configured in `mcp.json` are loaded asynchronously in the background, never blocking the main loop — chat is usable immediately. diff --git a/docs/zh/tools/mcp.mdx b/docs/zh/tools/mcp.mdx index 8b7670c1..c7479a98 100644 --- a/docs/zh/tools/mcp.mdx +++ b/docs/zh/tools/mcp.mdx @@ -37,6 +37,7 @@ Docker 部署时，官方 `docker-compose.yml` 已经把宿主机 `./cow` 挂载 | `url` | SSE / Streamable HTTP | 远程端点 URL（与 `command` 二选一） | | `type` | 远程 | 远程传输类型，可选 `sse` 或 `streamable-http`，默认 `sse` | | `headers` | 否 | 远程请求附加 HTTP 头（如 `Authorization`），仅 Streamable HTTP 使用 | +| `scope` | 否 | OAuth 授权范围，仅需要 OAuth 授权的远程 server 使用（可选） | | `disabled` | 否 | `true` 时跳过该 server，便于临时关闭 | ### 完整示例 @@ -79,6 +80,27 @@ Agent 会： 1. 访问 MCP 配置文件，合并新 server 配置，保留已有项 2. 自动重载增量的 MCP Server，下一次对话即可使用相应 Tools +## 网页授权（OAuth） + +部分远程 MCP需要 OAuth 网页授权，直接配置会返回 `401`。CowAgent 内置标准 OAuth 流程，**无需手填 token**，正常配置即可，例如： + +```json +{ + "mcpServers": { + "xmind": { + "type": "streamable-http", + "url": "https://app.xmind.com/api/mcp" + } + } +} +``` + +server 首次加载遇到 `401` 时会自动发起授权：本机运行会**自动打开浏览器**，服务器部署则把**授权链接打印到日志**，复制到浏览器打开。授权同意后即完成，该 server 随即上线，令牌过期自动刷新，无需重复授权。 + +- **依赖 Web 服务**：授权回调由 Web 控制台（默认端口 `9899`）接收，需保证 Web channel 正在运行。 +- **凭证存储**：令牌持久化在 `~/.cow/mcp_oauth.json`，重启后复用。 +- **回调地址**：默认 `http://127.0.0.1:9899/mcp/oauth/callback`；若部署在服务器、授权浏览器在另一台设备，在 `config.json` 设置 `mcp_oauth_redirect_base`（如 `http://你的IP:9899`）即可。 + ## 工作方式 - 启动时**异步加载**：`mcp.json` 中配置的所有 server 会在后台异步加载，不阻塞主流程，对话可以立刻使用