mcp server auth token

python/helpers/settings.py

@@ -1,3 +1,5 @@
+import base64
+import hashlib
 import json
 import os
 import re
@@ -67,6 +69,7 @@ class Settings(TypedDict):
     mcp_client_init_timeout: int
     mcp_client_tool_timeout: int
     mcp_server_enabled: bool
+    mcp_server_token: str
 
 
 class PartialSettings(Settings, total=False):
@@ -748,6 +751,17 @@ def convert_out(settings: Settings) -> SettingsOutput:
         }
     )
 
+    mcp_server_fields.append(
+        {
+            "id": "mcp_server_token",
+            "title": "MCP Server Token",
+            "description": "Token for MCP server authentication.",
+            "type": "text",
+            "hidden": True,
+            "value": settings["mcp_server_token"],
+        }
+    )
+
     mcp_server_section: SettingsSection = {
         "id": "mcp_server",
         "title": "A0 MCP Server",
@@ -845,6 +859,9 @@ def normalize_settings(settings: Settings) -> Settings:
             except (ValueError, TypeError):
                 copy[key] = value  # make default instead
 
+    # mcp server token is set automatically
+    copy["mcp_server_token"] = create_token()
+
     return copy
 
 
@@ -939,6 +956,7 @@ def get_default_settings() -> Settings:
         mcp_client_init_timeout=5,
         mcp_client_tool_timeout=120,
         mcp_server_enabled=False,
+        mcp_server_token=create_token(),
     )
 
 
@@ -1075,3 +1093,15 @@ def get_runtime_config(set: Settings):
             "code_exec_http_port": set["rfc_port_http"],
             "code_exec_ssh_user": "root",
         }
+
+
+def create_token() -> str:
+    username = dotenv.get_dotenv_value(dotenv.KEY_AUTH_LOGIN) or ""
+    password = dotenv.get_dotenv_value(dotenv.KEY_AUTH_PASSWORD) or ""
+    if not username or not password:
+        return "0"
+    # use base64 encoding for a more compact token with alphanumeric chars
+    hash_bytes = hashlib.sha256(f"{username}:{password}".encode()).digest()
+    # encode as base64 and remove any non-alphanumeric chars (like +, /, =)
+    b64_token = base64.urlsafe_b64encode(hash_bytes).decode().replace('=', '')
+    return b64_token[:16]

run_ui.py

@@ -240,19 +240,46 @@ def run():
 
         
     # define a Starlette-compatible middleware handler
+    from starlette.requests import Request
+    import re
 
-
-    async def mcp_middleware(request, call_next):
-        set = settings.get_settings()
-        if not set["mcp_server_enabled"]:
-            # raise a proper Starlette HTTPException with a clear message
+    async def mcp_middleware(request: Request, call_next):
+        
+        # check if MCP server is enabled
+        cfg = settings.get_settings()
+        if not cfg["mcp_server_enabled"]:
             PrintStyle.error("[MCP] Access denied: MCP server is disabled in settings.")
-            raise StarletteHTTPException(status_code=403, detail="MCP server is disabled in settings.")
+            raise StarletteHTTPException(status_code=403,
+                                        detail="MCP server is disabled in settings.")
+
+        # get auth token from path
+        full_path = request.url.path
+        if not full_path.startswith("/mcp/t-"):
+            raise StarletteHTTPException(status_code=401,
+                                        detail="Missing token.")            
+        pattern = r'^/mcp/t-([^/]+)/(.+)$'
+        match = re.match(pattern, full_path)
+        if not match:
+            raise StarletteHTTPException(status_code=401,
+                                        detail="Missing token.")            
+        token = match.group(1)     
+        remainder = match.group(2)  
+        
+        # validate token
+        if token != cfg["mcp_server_token"]:
+            raise StarletteHTTPException(status_code=401,
+                                        detail="Invalid token.")
+
+        # rewrite path to standard MCP path and continue
+        new_path = f"/mcp/{remainder}"
+        request.scope["path"] = new_path
+        request.scope["raw_path"] = new_path.encode()
+        # request.state.token = token
+
         return await call_next(request)
 
-    mcp_middlewares = [
-        Middleware(BaseHTTPMiddleware, dispatch=mcp_middleware)
-    ]
+
+    mcp_middlewares = [Middleware(BaseHTTPMiddleware, dispatch=mcp_middleware)]
 
     mcp_app = create_sse_app(
         server=mcp_server_instance,
@@ -269,7 +296,7 @@ def run():
     app = DispatcherMiddleware(webapp, {
         "/mcp": ASGIMiddleware(app=mcp_app), # type: ignore
     }) # type: ignore
-    PrintStyle().debug("Registered middleware for MCP")
+    PrintStyle().debug("Registered middleware for MCP and MCP token")
 
     try:
         PrintStyle().debug(f"Starting server at {host}:{port}...")

webui/components/settings/mcp/server/example.html

@@ -9,19 +9,20 @@
     <div x-data>
         <p>Agent Zero MCP Server is an SSE MCP running on the same URL and port as the Web UI + /mcp/sse path.</p>
         <p>The same applies if you run A0 on a public URL using a tunnel.</p>
-        
+
         <h3>Example MCP Server Configuration JSON</h3>
         <div id="mcp-server-example"></div>
 
         <script>
             setTimeout(() => {
                 const url = window.location.origin;
+                const token = settingsModalProxy.settings.sections.filter(x => x.id == "mcp_server")[0].fields.filter(x => x.id == "mcp_server_token")[0].value;
                 const jsonExample = JSON.stringify({
                     "mcpServers":
                     {
                         "agent-zero": {
                             "type": "sse",
-                            "serverUrl": `${url}/mcp/sse`
+                            "serverUrl": `${url}/mcp/t-${token}/sse`
                         }
                     }
                 }, null, 2);