feat(auth): implement RAM signature authentication for agentrun-data endpoints

adds comprehensive RAM signature authentication support for both OpenAPI and MCP tool invocations. introduces automatic URL rewriting to -ram endpoints when targeting agentrun-data domains, dynamic signature generation per request, and proper header injection. enhances error handling with detailed logging for HTTP status failures and removes deprecated debug functions.

The implementation includes:
- httpx Auth handler for dynamic RAM signature calculation per request
- automatic URL rewriting from agentrun-data to -ram endpoints
- enhanced MCP session with RAM authentication support
- improved OpenAPI tool invocation with RAM signature injection
- better error logging for failed requests
- conditional intranet URL usage based on FC_REGION environment variable
- unified MCP URL resolution with fallback mechanisms

BREAKING CHANGE: replaces assertion-based error handling with ValueError exceptions in MCP URL resolution

feat(auth): 为 agentrun-data 端点实现 RAM 签名认证

为 OpenAPI 和 MCP 工具调用添加了全面的 RAM 签名认证支持。引入了自动 URL 重写为 -ram 端点、动态签名生成和适当的头信息注入。增强了错误处理并提供详细的 HTTP 状态失败日志，移除了已弃用的调试函数。

实现包括：
- 用于每个请求动态计算 RAM 签名的 httpx Auth 处理器
- 自动将 agentrun-data 重写为 -ram 端点的 URL
- 支持 RAM 认证的增强 MCP 会话
- 具有 RAM 签名注入的改进 OpenAPI 工具调用
- 针对失败请求的改进错误日志
- 基于 FC_REGION 环境变量的条件内网 URL 使用
- 具有回退机制的统一 MCP URL 解析

重大变更：将 MCP URL 解析中的断言错误处理替换为 ValueError 异常

BREAKING CHANGE: replaces assertion-based error handling with ValueError exceptions in MCP URL resolution

重大变更: 在 MCP URL 解析中将基于断言的错误处理替换为 ValueError 异常

Change-Id: I415cb816f44f6eca78b7165b96b1fce8fd7d3404
Co-authored-by: Copilot <copilot@github.com>
Signed-off-by: OhYee <oyohyee@oyohyee.com>

Author: OhYee

agentrun/toolset/__toolset_async_template.py

@@ -94,11 +94,15 @@ def _get_openapi_auth_defaults(
         return headers, query
 
     def _get_openapi_base_url(self) -> Optional[str]:
-        intranet_url: Optional[str] = pydash.get(
-            self, "status.outputs.urls.intranet_url", None
-        )
-        if intranet_url:
-            return intranet_url
+        import os
+
+        fc_region = os.getenv("FC_REGION")
+        if fc_region:
+            intranet_url: Optional[str] = pydash.get(
+                self, "status.outputs.urls.intranet_url", None
+            )
+            if intranet_url:
+                return intranet_url
 
         return pydash.get(self, "status.outputs.urls.internet_url", None)
 
@@ -156,6 +160,25 @@ async def call_tool_async(
         logger.debug("invoke tool %s got result %s", name, result)
         return result
 
+    def _get_mcp_url(self) -> str:
+        """获取 MCP 工具的最佳 URL / Get the best URL for MCP tool
+
+        优先使用 agentrun-data 代理入口（支持 RAM 签名认证），
+        回退到 mcp_server_config.url（直连）。
+        Priority: agentrun-data proxy endpoint (with RAM auth) > mcp_server_config.url (direct).
+        """
+        proxy_url = self._get_openapi_base_url()
+        if proxy_url:
+            return proxy_url
+
+        mcp_server_config: MCPServerConfig = pydash.get(
+            self, "status.outputs.mcp_server_config", None
+        )
+        if mcp_server_config and mcp_server_config.url:
+            return mcp_server_config.url
+
+        raise ValueError("MCP server URL is missing.")
+
     def to_apiset(self, config: Optional[Config] = None):
         """将 ToolSet 转换为统一的 ApiSet 对象
 
@@ -170,16 +193,16 @@ def to_apiset(self, config: Optional[Config] = None):
             mcp_server_config: MCPServerConfig = pydash.get(
                 self, "status.outputs.mcp_server_config", None
             )
-            assert (
-                mcp_server_config.url is not None
-            ), "MCP server URL is missing."
 
-            cfg = Config.with_configs(
-                config, Config(headers=mcp_server_config.headers)
+            mcp_url = self._get_mcp_url()
+
+            mcp_headers = (
+                mcp_server_config.headers if mcp_server_config else None
             )
+            cfg = Config.with_configs(config, Config(headers=mcp_headers))
 
             mcp_client = MCPToolSet(
-                url=mcp_server_config.url,
+                url=mcp_url,
                 config=cfg,
             )
 

agentrun/toolset/api/mcp.py

@@ -4,10 +4,88 @@
 Handles tool invocations for MCP (Model Context Protocol).
 """
 
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Generator, Optional
+from urllib.parse import urlparse, urlunparse
+
+import httpx
 
 from agentrun.utils.config import Config
 from agentrun.utils.log import logger
+from agentrun.utils.ram_signature import get_agentrun_signed_headers
+
+
+class _AgentrunRamAuth(httpx.Auth):
+    """httpx Auth handler：为每次请求动态生成 RAM 签名。
+
+    SSE 场景下同一个 httpx.AsyncClient 会发出 GET（SSE 连接）和
+    POST（消息发送）请求，URL / method / body 各不相同，因此必须
+    per-request 计算签名，不能在 client 初始化时一次性设置 headers。
+    """
+
+    def __init__(
+        self,
+        access_key_id: str,
+        access_key_secret: str,
+        region: str,
+        security_token: Optional[str] = None,
+    ):
+        self._ak = access_key_id
+        self._sk = access_key_secret
+        self._region = region
+        self._security_token = security_token
+
+    def auth_flow(
+        self, request: httpx.Request
+    ) -> Generator[httpx.Request, httpx.Response, None]:
+        url = str(request.url)
+        method = request.method
+
+        body: Optional[bytes] = None
+        if request.content:
+            body = request.content
+
+        content_type: Optional[str] = request.headers.get("content-type")
+
+        try:
+            signed = get_agentrun_signed_headers(
+                url=url,
+                method=method,
+                access_key_id=self._ak,
+                access_key_secret=self._sk,
+                security_token=self._security_token,
+                region=self._region,
+                product="agentrun",
+                body=body,
+                content_type=content_type,
+            )
+            for k, v in signed.items():
+                request.headers[k] = v
+            logger.debug(
+                "applied RAM signature for MCP %s request to %s",
+                method,
+                url[:80] + ("..." if len(url) > 80 else ""),
+            )
+        except ValueError as e:
+            logger.warning("RAM signing skipped for MCP request: %s", e)
+
+        yield request
+
+
+def _rewrite_to_ram_url(url: str) -> str:
+    """将 agentrun-data 域名改写为 -ram 端点。"""
+    parsed = urlparse(url)
+    parts = parsed.netloc.split(".", 1)
+    if len(parts) == 2:
+        ram_netloc = parts[0] + "-ram." + parts[1]
+        return urlunparse((
+            parsed.scheme,
+            ram_netloc,
+            parsed.path or "",
+            parsed.params,
+            parsed.query,
+            parsed.fragment,
+        ))
+    return url
 
 
 class MCPSession:
@@ -16,14 +94,46 @@ def __init__(self, url: str, config: Optional[Config] = None):
         self.url = url
         self.config = Config.with_configs(config)
 
+    def _build_ram_auth(self, url: str) -> tuple:
+        """当目标是 agentrun-data 域名时，改写 URL 并返回 httpx Auth handler。
+
+        Returns:
+            (rewritten_url, auth_or_none)
+        """
+        parsed = urlparse(url)
+        if ".agentrun-data." not in (parsed.netloc or ""):
+            return url, None
+
+        cfg = self.config
+        ak = cfg.get_access_key_id()
+        sk = cfg.get_access_key_secret()
+        if not ak or not sk:
+            return url, None
+
+        url = _rewrite_to_ram_url(url)
+
+        auth = _AgentrunRamAuth(
+            access_key_id=ak,
+            access_key_secret=sk,
+            region=cfg.get_region_id(),
+            security_token=cfg.get_security_token() or None,
+        )
+        return url, auth
+
     async def __aenter__(self):
         from mcp import ClientSession
         from mcp.client.sse import sse_client
 
         timeout = self.config.get_timeout()
+        headers = self.config.get_headers()
+        url = self.url
+
+        url, auth = self._build_ram_auth(url)
+
         self.client = sse_client(
-            url=self.url,
-            headers=self.config.get_headers(),
+            url=url,
+            headers=headers,
+            auth=auth,
             timeout=timeout if timeout else 60,
         )
         read, write = await self.client.__aenter__()

agentrun/toolset/api/openapi.py

@@ -7,6 +7,7 @@
 from copy import deepcopy
 import json
 from typing import Any, Dict, List, Optional, Tuple, Union
+from urllib.parse import urlparse, urlunparse
 
 import httpx
 from pydash import get as pg
@@ -15,6 +16,7 @@
 from agentrun.utils.config import Config
 from agentrun.utils.log import logger
 from agentrun.utils.model import BaseModel
+from agentrun.utils.ram_signature import get_agentrun_signed_headers
 
 from ..model import ToolInfo, ToolSchema
 
@@ -830,11 +832,21 @@ def invoke_tool(
             name, arguments, config
         )
 
-        print(request_kwargs)
         with httpx.Client(timeout=timeout) as client:
             response = client.request(**request_kwargs)
             if raise_for_status:
-                response.raise_for_status()
+                try:
+                    response.raise_for_status()
+                except httpx.HTTPStatusError:
+                    logger.error(
+                        "OpenAPI tool request failed: status=%s url=%s "
+                        "response_headers=%s response_body=%s",
+                        response.status_code,
+                        response.request.url,
+                        dict(response.headers),
+                        response.text[:2000],
+                    )
+                    raise
             return self._format_response(response)
 
     async def invoke_tool_async(
@@ -850,7 +862,18 @@ async def invoke_tool_async(
         async with httpx.AsyncClient(timeout=timeout) as client:
             response = await client.request(**request_kwargs)
             if raise_for_status:
-                response.raise_for_status()
+                try:
+                    response.raise_for_status()
+                except httpx.HTTPStatusError:
+                    logger.error(
+                        "OpenAPI tool request failed: status=%s url=%s "
+                        "response_headers=%s response_body=%s",
+                        response.status_code,
+                        response.request.url,
+                        dict(response.headers),
+                        response.text[:2000],
+                    )
+                    raise
             return self._format_response(response)
 
     def _load_schema(self, schema: Any) -> Dict[str, Any]:
@@ -945,7 +968,6 @@ def _walk(node: Any):
 
     def _extract_base_url(self, schema: Dict[str, Any]) -> Optional[str]:
         servers = schema.get("servers") or []
-        print("======", servers)
         return self._pick_server_url(servers)
 
     def _convert_to_native(self, value: Any) -> Any:
@@ -1177,8 +1199,66 @@ def _prepare_request(
                 args,
             )
 
+        self._apply_ram_auth(request_kwargs, combined_config)
+
         return request_kwargs, timeout, raise_for_status
 
+    def _apply_ram_auth(
+        self,
+        request_kwargs: Dict[str, Any],
+        config: Optional[Config],
+    ) -> None:
+        """当目标是 agentrun-data 域名时，自动注入 RAM 签名鉴权 headers 并改写为 -ram 端点。"""
+        url = request_kwargs.get("url", "")
+        parsed = urlparse(url)
+        if ".agentrun-data." not in (parsed.netloc or ""):
+            return
+
+        cfg = Config.with_configs(config)
+        ak = cfg.get_access_key_id()
+        sk = cfg.get_access_key_secret()
+        if not ak or not sk:
+            return
+
+        parts = parsed.netloc.split(".", 1)
+        if len(parts) == 2:
+            ram_netloc = parts[0] + "-ram." + parts[1]
+            url = urlunparse((
+                parsed.scheme,
+                ram_netloc,
+                parsed.path or "",
+                parsed.params,
+                parsed.query,
+                parsed.fragment,
+            ))
+            request_kwargs["url"] = url
+
+        method = request_kwargs.get("method", "GET")
+        body_bytes: Optional[bytes] = None
+        json_body = request_kwargs.get("json")
+        if json_body is not None:
+            body_bytes = json.dumps(json_body).encode("utf-8")
+
+        try:
+            signed = get_agentrun_signed_headers(
+                url=url,
+                method=method,
+                access_key_id=ak,
+                access_key_secret=sk,
+                security_token=cfg.get_security_token() or None,
+                region=cfg.get_region_id(),
+                product="agentrun",
+                body=body_bytes,
+            )
+            existing_headers: Dict[str, str] = request_kwargs.get("headers", {})
+            request_kwargs["headers"] = {**signed, **existing_headers}
+            logger.debug(
+                "applied RAM signature for OpenAPI tool request to %s",
+                url[:80] + "..." if len(url) > 80 else url,
+            )
+        except ValueError as e:
+            logger.warning("RAM signing skipped for OpenAPI tool: %s", e)
+
     def _format_response(self, response: httpx.Response) -> Dict[str, Any]:
         try:
             body = response.json()