Merge pull request #2950 from zhayujie/feat-mcp-oauth

feat(mcp): auto OAuth authorization for remote MCP servers

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:<web_port>). 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:

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

agent/tools/tool_manager.py

@@ -466,17 +466,26 @@ 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", "<unnamed>")
                 try:
                     client = McpClient(cfg)
                     if not client.initialize():
+                        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"
@@ -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)

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 (
+                "<!doctype html><html><head><meta charset='utf-8'>"
+                "<meta name='viewport' content='width=device-width,initial-scale=1'>"
+                f"<title>{title}</title></head>"
+                "<body style='font-family:-apple-system,Segoe UI,Roboto,sans-serif;"
+                "max-width:520px;margin:64px auto;padding:0 20px;text-align:center;color:#1f2328'>"
+                f"<h2>{title}</h2><p style='color:#57606a'>{message}</p></body></html>"
+            )
+
+        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')

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

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 はバックグラウンドで非同期に読み込まれ、メインループをブロックしません。会話はすぐに開始できます

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.

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 会在后台异步加载，不阻塞主流程，对话可以立刻使用