feat: add Config.from_request_headers to extract FC temporary credentials from HTTP headers

FC platform injects fresh STS credentials via x-fc-access-key-id, x-fc-access-key-secret,
and x-fc-security-token headers on every request. This method extracts them so long-running
agent services can use per-request credentials instead of potentially expired env vars.

- Add Config.from_request_headers() classmethod supporting AgentRequest and Starlette Request
- Add unit tests (8 cases) and e2e tests (5 cases covering OpenAI/AG-UI protocols)
- Add example: server_with_header_credentials.py

Change-Id: I59a5844735797a26a30fa04a7859e2e99d216d17
Co-developed-by: Claude <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: congxiao-wxx

agentrun/utils/config.py

@@ -5,7 +5,7 @@
 """
 
 import os
-from typing import Dict, Optional
+from typing import Any, Dict, Optional
 
 from dotenv import load_dotenv
 
@@ -154,6 +154,40 @@ def __init__(
         self._bailian_endpoint = bailian_endpoint
         self._headers = headers or {}
 
+    @classmethod
+    def from_request_headers(cls, request: Any) -> "Config":
+        """从 HTTP 请求头提取临时凭证构造 Config / Create Config from HTTP request headers
+
+        支持传入 AgentRequest（自动取 raw_request）或 Starlette Request。
+        Accepts AgentRequest (auto-extracts raw_request) or Starlette Request.
+
+        提取以下 header:
+        Extracts the following headers:
+        - x-fc-access-key-id → access_key_id
+        - x-fc-access-key-secret → access_key_secret
+        - x-fc-security-token → security_token
+
+        Args:
+            request: AgentRequest 或 Starlette Request 对象
+                     AgentRequest or Starlette Request object
+
+        Returns:
+            Config: 包含请求头中凭证信息的配置对象
+                    Config with credentials from request headers
+
+        Note:
+            如果需要与其他 Config 合并,请将此方法返回的 Config 放在 with_configs 的最后参数,
+            以确保 header 凭证不被覆盖:
+            ``Config.with_configs(base_config, Config.from_request_headers(request))``
+        """
+        raw = getattr(request, "raw_request", None) or request
+        headers = getattr(raw, "headers", {})
+        return cls(
+            access_key_id=headers.get("x-fc-access-key-id", ""),
+            access_key_secret=headers.get("x-fc-access-key-secret", ""),
+            security_token=headers.get("x-fc-security-token", ""),
+        )
+
     @classmethod
     def with_configs(cls, *configs: Optional["Config"]) -> "Config":
         return cls().update(*configs)

examples/server_with_header_credentials.py

@@ -0,0 +1,68 @@
+"""使用 HTTP 请求头临时凭证的 Agent Server 示例
+
+本示例演示如何在 AgentRunServer 中从每次请求的 HTTP header 获取临时凭证，
+替代环境变量中可能过期的 token，确保 Agent 调用知识库等资源时凭证始终有效。
+
+适用场景：
+- 部署在函数计算（FC）上的 Agent，FC 平台通过请求头注入临时凭证
+- 需要长期运行的 Agent 服务，环境变量中的 STS Token 会过期
+
+FC 平台注入的请求头：
+- x-fc-access-key-id: 临时 AccessKey ID
+- x-fc-access-key-secret: 临时 AccessKey Secret
+- x-fc-security-token: 临时 Security Token
+
+启动服务后测试：
+    curl http://127.0.0.1:9000/openai/v1/chat/completions -X POST \\
+        -H "Content-Type: application/json" \\
+        -H "x-fc-access-key-id: <your-ak>" \\
+        -H "x-fc-access-key-secret: <your-sk>" \\
+        -H "x-fc-security-token: <your-token>" \\
+        -d '{"messages": [{"role": "user", "content": "什么是Serverless?"}], "stream": true}'
+"""
+
+import os
+
+from agentrun.knowledgebase import KnowledgeBase
+from agentrun.server import AgentRequest, AgentRunServer
+from agentrun.server.model import ServerConfig
+from agentrun.utils.config import Config
+from agentrun.utils.log import logger
+
+KNOWLEDGE_BASE_NAME = os.getenv("AGENTRUN_KNOWLEDGE_BASE", "my-knowledge-base")
+
+
+async def invoke_agent(request: AgentRequest):
+    config = Config.from_request_headers(request)
+
+    user_query = request.messages[-1].content if request.messages else ""
+    if not user_query:
+        yield "请输入您的问题。"
+        return
+
+    try:
+        kb = KnowledgeBase.get_by_name(KNOWLEDGE_BASE_NAME, config=config)
+        result = kb.retrieve(query=user_query, config=config)
+
+        nodes = result.get("nodes", result.get("data", []))
+        if not nodes:
+            yield f"未在知识库 {KNOWLEDGE_BASE_NAME} 中找到相关内容。"
+            return
+
+        chunks = [
+            node.get("content", node.get("text", ""))
+            for node in nodes
+            if node.get("content") or node.get("text")
+        ]
+        context = "\n---\n".join(chunks)
+        yield f"根据知识库检索结果：\n\n{context}"
+
+    except Exception as e:
+        logger.error("知识库检索失败: %s", e)
+        yield f"知识库检索失败: {e}"
+
+
+AgentRunServer(
+    invoke_agent=invoke_agent,
+    config=ServerConfig(cors_origins=["*"]),
+).start(port=9000)

tests/e2e/test_header_credentials.py

@@ -0,0 +1,167 @@
+"""
+E2E 测试: Config.from_request_headers 通过真实 HTTP 请求提取临时凭证
+
+测试覆盖:
+- 通过 OpenAI 协议发送 x-fc-* header，invoke_agent 中使用 Config.from_request_headers 提取凭证
+- 通过 AG-UI 协议发送 x-fc-* header，验证同样可以提取
+- 部分 header 缺失时对应字段为空字符串
+- 无 x-fc-* header 时所有字段为空字符串
+"""
+
+from agentrun.server import AgentRequest, AgentRunServer
+from agentrun.utils.config import Config
+
+
+def _make_client(invoke_agent):
+    server = AgentRunServer(invoke_agent=invoke_agent)
+    app = server.as_fastapi_app()
+    from fastapi.testclient import TestClient
+
+    return TestClient(app)
+
+
+class TestHeaderCredentials:
+    """通过真实 HTTP 请求验证 Config.from_request_headers 的 E2E 行为"""
+
+    def test_openai_full_headers(self):
+        """OpenAI 协议: 三个 x-fc-* header 都存在时正确提取"""
+        captured = {}
+
+        async def invoke_agent(request: AgentRequest):
+            config = Config.from_request_headers(request)
+            captured["ak"] = config.get_access_key_id()
+            captured["sk"] = config.get_access_key_secret()
+            captured["token"] = config.get_security_token()
+            yield "ok"
+
+        client = _make_client(invoke_agent)
+        response = client.post(
+            "/openai/v1/chat/completions",
+            json={
+                "messages": [{"role": "user", "content": "test"}],
+                "stream": True,
+            },
+            headers={
+                "x-fc-access-key-id": "ak-e2e-test",
+                "x-fc-access-key-secret": "sk-e2e-test",
+                "x-fc-security-token": "token-e2e-test",
+            },
+        )
+        assert response.status_code == 200
+        assert captured["ak"] == "ak-e2e-test"
+        assert captured["sk"] == "sk-e2e-test"
+        assert captured["token"] == "token-e2e-test"
+
+    def test_agui_full_headers(self):
+        """AG-UI 协议: 三个 x-fc-* header 都存在时正确提取"""
+        captured = {}
+
+        async def invoke_agent(request: AgentRequest):
+            config = Config.from_request_headers(request)
+            captured["ak"] = config.get_access_key_id()
+            captured["sk"] = config.get_access_key_secret()
+            captured["token"] = config.get_security_token()
+            yield "ok"
+
+        client = _make_client(invoke_agent)
+        response = client.post(
+            "/ag-ui/agent",
+            json={
+                "messages": [{"role": "user", "content": "test"}],
+            },
+            headers={
+                "x-fc-access-key-id": "ak-agui-test",
+                "x-fc-access-key-secret": "sk-agui-test",
+                "x-fc-security-token": "token-agui-test",
+            },
+        )
+        assert response.status_code == 200
+        assert captured["ak"] == "ak-agui-test"
+        assert captured["sk"] == "sk-agui-test"
+        assert captured["token"] == "token-agui-test"
+
+    def test_partial_headers(self):
+        """部分 x-fc-* header 缺失时，缺失字段为空字符串"""
+        captured = {}
+
+        async def invoke_agent(request: AgentRequest):
+            config = Config.from_request_headers(request)
+            captured["ak"] = config.get_access_key_id()
+            captured["sk"] = config.get_access_key_secret()
+            captured["token"] = config.get_security_token()
+            yield "ok"
+
+        client = _make_client(invoke_agent)
+        response = client.post(
+            "/openai/v1/chat/completions",
+            json={
+                "messages": [{"role": "user", "content": "test"}],
+                "stream": True,
+            },
+            headers={
+                "x-fc-access-key-id": "ak-partial",
+            },
+        )
+        assert response.status_code == 200
+        assert captured["ak"] == "ak-partial"
+        assert captured["sk"] == ""
+        assert captured["token"] == ""
+
+    def test_no_fc_headers(self):
+        """无 x-fc-* header 时所有字段为空字符串"""
+        captured = {}
+
+        async def invoke_agent(request: AgentRequest):
+            config = Config.from_request_headers(request)
+            captured["ak"] = config.get_access_key_id()
+            captured["sk"] = config.get_access_key_secret()
+            captured["token"] = config.get_security_token()
+            yield "ok"
+
+        client = _make_client(invoke_agent)
+        response = client.post(
+            "/openai/v1/chat/completions",
+            json={
+                "messages": [{"role": "user", "content": "test"}],
+                "stream": True,
+            },
+        )
+        assert response.status_code == 200
+        assert captured["ak"] == ""
+        assert captured["sk"] == ""
+        assert captured["token"] == ""
+
+    def test_config_merge_with_base(self):
+        """从 header 提取的 Config 与 base Config 合并时 header 凭证优先"""
+        captured = {}
+
+        async def invoke_agent(request: AgentRequest):
+            base = Config(
+                access_key_id="base-ak",
+                access_key_secret="base-sk",
+                security_token="base-token",
+            )
+            header_config = Config.from_request_headers(request)
+            merged = Config.with_configs(base, header_config)
+            captured["ak"] = merged.get_access_key_id()
+            captured["sk"] = merged.get_access_key_secret()
+            captured["token"] = merged.get_security_token()
+            yield "ok"
+
+        client = _make_client(invoke_agent)
+        response = client.post(
+            "/openai/v1/chat/completions",
+            json={
+                "messages": [{"role": "user", "content": "test"}],
+                "stream": True,
+            },
+            headers={
+                "x-fc-access-key-id": "header-ak",
+                "x-fc-access-key-secret": "header-sk",
+                "x-fc-security-token": "header-token",
+            },
+        )
+        assert response.status_code == 200
+        assert captured["ak"] == "header-ak"
+        assert captured["sk"] == "header-sk"
+        assert captured["token"] == "header-token"

tests/unittests/test_config.py

@@ -0,0 +1,105 @@
+"""Config.from_request_headers 单元测试"""
+
+from agentrun.utils.config import Config
+
+
+class FakeRequest:
+    """模拟 Starlette Request"""
+
+    def __init__(self, headers: dict):
+        self.headers = headers
+
+
+class FakeAgentRequest:
+    """模拟 AgentRequest（含 raw_request）"""
+
+    def __init__(self, headers: dict):
+        self.raw_request = FakeRequest(headers)
+
+
+class TestFromRequestHeaders:
+
+    def test_full_headers(self):
+        request = FakeRequest({
+            "x-fc-access-key-id": "ak-123",
+            "x-fc-access-key-secret": "sk-456",
+            "x-fc-security-token": "token-789",
+        })
+        config = Config.from_request_headers(request)
+
+        assert config.get_access_key_id() == "ak-123"
+        assert config.get_access_key_secret() == "sk-456"
+        assert config.get_security_token() == "token-789"
+
+    def test_partial_headers_missing_token(self):
+        request = FakeRequest({
+            "x-fc-access-key-id": "ak-123",
+            "x-fc-access-key-secret": "sk-456",
+        })
+        config = Config.from_request_headers(request)
+
+        assert config.get_access_key_id() == "ak-123"
+        assert config.get_access_key_secret() == "sk-456"
+        assert config.get_security_token() == ""
+
+    def test_partial_headers_only_key_id(self):
+        request = FakeRequest({
+            "x-fc-access-key-id": "ak-only",
+        })
+        config = Config.from_request_headers(request)
+
+        assert config.get_access_key_id() == "ak-only"
+        assert config.get_access_key_secret() == ""
+        assert config.get_security_token() == ""
+
+    def test_empty_headers(self):
+        request = FakeRequest({})
+        config = Config.from_request_headers(request)
+
+        assert config.get_access_key_id() == ""
+        assert config.get_access_key_secret() == ""
+        assert config.get_security_token() == ""
+
+    def test_agent_request_unwrap(self):
+        agent_req = FakeAgentRequest({
+            "x-fc-access-key-id": "ak-from-agent",
+            "x-fc-access-key-secret": "sk-from-agent",
+            "x-fc-security-token": "token-from-agent",
+        })
+        config = Config.from_request_headers(agent_req)
+
+        assert config.get_access_key_id() == "ak-from-agent"
+        assert config.get_access_key_secret() == "sk-from-agent"
+        assert config.get_security_token() == "token-from-agent"
+
+    def test_config_usable_as_resource_param(self):
+        config = Config.from_request_headers(FakeRequest({
+            "x-fc-access-key-id": "ak-new",
+            "x-fc-access-key-secret": "sk-new",
+            "x-fc-security-token": "token-new",
+        }))
+        assert config.get_access_key_id() == "ak-new"
+        assert config.get_access_key_secret() == "sk-new"
+        assert config.get_security_token() == "token-new"
+        assert config.get_region_id() == "cn-hangzhou"
+
+    def test_agent_request_with_none_raw_request(self):
+        agent_req = FakeAgentRequest({})
+        agent_req.raw_request = None
+        config = Config.from_request_headers(agent_req)
+        assert config.get_access_key_id() == ""
+        assert config.get_access_key_secret() == ""
+        assert config.get_security_token() == ""
+
+    def test_extra_headers_ignored(self):
+        request = FakeRequest({
+            "x-fc-access-key-id": "ak-123",
+            "x-fc-access-key-secret": "sk-456",
+            "x-fc-security-token": "token-789",
+            "authorization": "Bearer xxx",
+            "content-type": "application/json",
+        })
+        config = Config.from_request_headers(request)
+        assert config.get_access_key_id() == "ak-123"
+        assert config.get_access_key_secret() == "sk-456"
+        assert config.get_security_token() == "token-789"