feat: claude_agent_sdk provider — 节点可走个人 Claude Pro/Max 订阅额度 (v0.4.0, PR #86)

This commit is contained in:
Simon Lin
2026-07-31 11:40:03 +12:00
parent d8c64507e0
commit 906c2baad3
12 changed files with 1202 additions and 15 deletions
+7
View File
@@ -13,3 +13,10 @@ OPENROUTER_API_KEY=
# 方便国内通过中转访问 Claude / OpenAI 等。Web UI 侧边栏也可直接填写。
# 例: BACKEND_URL=https://your-proxy.com/v1
BACKEND_URL=
# Claude Agent SDK provider(仅个人自用,可选依赖 [agentsdk])。
# 让 deep_thinking_llm 节点走你个人 Claude Pro/Max 订阅额度而非按 token 计费的 API。
# 用 `claude setup-token`(已登录 Pro/Max 账号)生成后填这里。
# ⚠️ 与 ANTHROPIC_API_KEY 不可共存:API key 优先级更高会悄悄走 API 计费,
# 启用该 provider 时若检测到 ANTHROPIC_API_KEY 会直接报错中止(护栏)。
CLAUDE_CODE_OAUTH_TOKEN=
+57
View File
@@ -6,6 +6,63 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
Breaking changes within the 0.x line are called out explicitly.
## [0.4.0] — 2026-07-31
新增:让节点走**个人 Claude Pro/Max 订阅额度**而非按 token 计费的 Anthropic API。
基于社区 PR [#86](https://github.com/simonlin1212/TradingAgents-astock/pull/86)
(感谢 [@kingxiaozhe](https://github.com/kingxiaozhe))。默认关闭,行为不变。
### 新增:`claude_agent_sdk` provider(可选依赖 `[agentsdk]`)
此前项目对 Claude 只有 `anthropic` provider=走 `ANTHROPIC_API_KEY` **按 token 计费**,
没有任何走订阅的通道。本版补上:经 Claude Agent SDK 调用本机已登录的 `claude`,
**消耗订阅额度,不产生 API 账单**。
- **两档覆盖**:`deep_think_provider_override` 只覆盖深度节点(Research / Portfolio
Manager);再加 `quick_think_provider_override` 则含 7 个工具分析师=全节点走订阅。
Web 侧栏三档(关闭 / 仅深度 / 所有节点)。
- **工具桥接**:分析师的 LangChain 工具桥接为 Agent SDK 进程内 MCP 工具,SDK 内部跑完
ReAct 循环返回最终报告,LangGraph 视为完成,无需改图。
- **启动护栏**:检测到 `ANTHROPIC_API_KEY` 与订阅覆盖共存时**直接报错中止**——
API key 优先级更高,留着它会悄悄走 API 计费。
- **撞额度自动降级**到配置的付费 provider(`agent_sdk_fallback_provider` / `_model`)。
- 仅供个人自用:消耗的是使用者自己账号的额度。
### 在 PR 基础上的四处改动
1. **模型默认值改用别名**。PR 默认 `claude-opus-4-8`,写死的完整 id 会随版本迭代过期
(仓库 `model_catalog` 的 anthropic 条目就已停在 `4-6`)。改用 `claude` CLI 的
`opus` / `sonnet` 别名——恒指向最新模型。完整 id 仍然支持。
quick 节点默认给 `sonnet` 而非 `opus`:节点数量多(7 分析师 + 辩手),
订阅按额度限流,全用 opus 很快撞上限。
2. **🔴 凭据失效不再静默降级到计费 provider**。原实现里认证失败会走
`_SDKResultError` → 落进 `_FALLBACK_ERRORS` → 降级到按 token 计费的 provider。
用户开订阅模式就是为了避免账单,token 一过期就悄悄开始计费——正是启动护栏 F-004
想防的事,只是从启动时挪到了运行中。新增 `_AuthError`(**刻意不在** `_FALLBACK_ERRORS`
里)+ `_looks_like_auth_failure()` 正向识别,报错里直接给出 `claude setup-token` 修复步骤。
> 实测发现有必要:OAuth token 过期时 SDK 把它翻译成 `Claude Code returned an
> error result: success`,`ResultMessage.subtype` 仍是 `"success"` 而 `is_error=True`,
> 真正原因只出现在 `api_retry` 事件与助手文本里。用户看到那句话完全无从下手。
3. **未采纳 PR 夹带的全局默认变更**:PR 把 `llm_provider` 默认从 `openai/gpt-5.4`
改成 `deepseek/deepseek-v4-pro`,那是贡献者个人偏好,会改变所有人的默认行为。
4. **未采纳 PR 的 `docs/codebase-context/` 与 `specs/`**(约 600 行贡献者自用的
spec-driven 脚手架文档)。另外 PR 的 `factory.py` 基于较旧的 main,整体取用会
**回退 v0.2.20 的 `openai_compatible` provider**——已改为只补 4 行路由,
测试 `test_openai_compatible_is_routed_to_openai_client` 当场抓到了这个回退。
### 依赖
`[agentsdk]` 链路为 `claude-agent-sdk → mcp → httpx2`,**不碰 httpx**,与 mootdx 的
`httpx<0.26` 无冲突(`uv lock --dry-run` 实测通过)。与 #87 中被移除的 `[google]`
情况不同,无需单开 venv。PR 注释里「会与 mootdx 冲突」的说法已过时(mcp 已迁到 httpx2),一并订正。
### 测试
`pytest tests/` **201 passed / 1 skipped / 45 subtests**。新增认证失败识别、
`_AuthError` 不参与降级、报错可操作性三组断言。端到端实跑验证链路可达
(本机 OAuth token 已过期,正确地报出可操作错误而非静默降级)。
## [0.3.1] — 2026-07-31
修三个静默失败 + 合并两个社区 PR。无破坏性变更。
+1 -1
View File
@@ -6,7 +6,7 @@
- **仓库**: https://github.com/simonlin1212/TradingAgents-astock
- **协议**: Apache 2.0
- **Python**: >=3.10
- **当前版本**: 0.3.1
- **当前版本**: 0.4.0
## 架构
+51 -1
View File
@@ -152,13 +152,18 @@ pip install -e .
# 如需使用 Google Gemini 模型(无 [google] extra,需显式装,见下方 FAQ):
pip install --no-deps "langchain-google-genai>=4.0.0"
pip install "google-genai>=1.53.0" "httpx>=0.28.1"
# 如需让节点走你个人 Claude Pro/Max 订阅额度而非 API 计费(可选):
pip install -e ".[agentsdk]"
```
> **装完即可用,无需 Docker。** 安装后直接跑 `streamlit run web/app.py`(Web UI)或 `tradingagents`(CLI)即可,详见下方「Web UI」「CLI 方式」两节。Docker 仅是可选的部署方式,本地开发不需要。
### 2. 配置 LLM
> **必须使用 API Key**,不能用 Claude/ChatGPT 订阅版。每次分析需 30-50 次 LLM 调用,只有 API 模式支持。
> **默认走 API Key 计费**。每次分析需 30-50 次 LLM 调用。
>
> **例外(v0.4.0 新增)**:装 `[agentsdk]` 后可让部分或全部节点经 Claude Agent SDK 走你**个人 Claude Pro/Max 订阅额度**,不产生 API 账单。见下方「用个人 Claude 订阅额度」。
在项目根目录创建 `.env` 文件,按你选择的供应商配置:
@@ -432,3 +437,48 @@ TradingAgents-Astock/
本项目是 TauricResearch/TradingAgents 的 fork,继承 Apache 2.0 许可证。详见 [NOTICE](./NOTICE)。
**作者:** Simon 林 · X [@linsizhen](https://x.com/linsizhen) · 邮箱:[simonlin0423@gmail.com](mailto:simonlin0423@gmail.com)
### 用个人 Claude 订阅额度(可选,v0.4.0 新增)
让节点经 Claude Agent SDK 走你**个人 Claude Pro/Max 订阅额度**,而不是按 token 计费的 Anthropic API。
> 与内置 `anthropic` provider 的区别:`anthropic` 走 `ANTHROPIC_API_KEY` = **按 token 计费**;本 provider 走本机已登录的 `claude` CLI = **消耗订阅额度,不产生 API 账单**。
>
> 仅供**个人自用**——它消耗的是你自己账号的订阅额度。把它做成给别人用的产品需要 Anthropic 授权,不在本项目范围内。
#### 1. 准备
```bash
pip install -e ".[agentsdk]"
# 本机 claude 已登录即可;headless / CI 环境需要显式 token:
claude setup-token # 输出的 token 设为 CLAUDE_CODE_OAUTH_TOKEN
# 关键:清掉 ANTHROPIC_API_KEY
unset ANTHROPIC_API_KEY
```
⚠️ **`ANTHROPIC_API_KEY` 优先级高于订阅 token**,留着它会悄悄走 API 计费。启动护栏检测到二者共存会**直接报错中止**,而不是让你事后看账单才发现。
#### 2. 开启
Web UI 侧栏「个人 Claude 订阅覆盖 (Agent SDK)」三档,或在 config 里设:
```python
config["deep_think_provider_override"] = "claude_agent_sdk" # 仅深度节点
config["quick_think_provider_override"] = "claude_agent_sdk" # 再加这条 = 全节点
config["agent_sdk_model"] = "opus" # 深度节点
config["agent_sdk_quick_model"] = "sonnet" # 分析师节点
```
**模型建议填别名 `opus` / `sonnet`**——`claude` CLI 的别名恒指向最新模型,写死 `claude-opus-4-8` 这类完整 id 会随版本迭代过期。完整 id 同样支持。
#### 3. 两条边界
- **额度而非 token**:订阅是按额度限流的。「所有节点」会把 7 个分析师 + 多空/交易员/风险辩手全压上去,跑几轮就可能撞上限——所以 `agent_sdk_quick_model` 默认给的是更省的 `sonnet`。撞额度会自动降级到你配的付费 provider(可用 `agent_sdk_fallback_provider` / `agent_sdk_fallback_model` 指定)。
- **凭据失效不降级**:OAuth token 过期时**直接报错中止**,不会静默降级到计费 provider——你开订阅模式就是为了避免账单,悄悄开始计费比报错更糟。报错里会给出 `claude setup-token` 的修复步骤。
#### 依赖说明
`[agentsdk]` 的依赖链是 `claude-agent-sdk → mcp → httpx2`,**不碰 httpx**,与 mootdx 的 `httpx<0.26` 无冲突(已 `uv lock` 实测)——和 #87 里被移除的 `[google]` 情况不同,不需要单开 venv。
+7 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "tradingagents-astock"
version = "0.3.1"
version = "0.4.0"
description = "A股多Agent投研框架 — 基于 TradingAgents 深度特化"
readme = "README.md"
requires-python = ">=3.10"
@@ -41,6 +41,12 @@ dependencies = [
"mootdx>=0.11.7",
]
[project.optional-dependencies]
# 走个人 Claude Pro/Max 订阅额度(而非按 token 计费的 Anthropic API)。
# 依赖链为 claude-agent-sdk → mcp → httpx2,**不碰 httpx**,故与 mootdx 的
# httpx<0.26 无冲突(已 uv lock 实测)——与 #87 移除的 [google] 情况不同。
agentsdk = ["claude-agent-sdk>=0.2.82"]
[project.urls]
Homepage = "https://github.com/simonlin1212/tradingagents-astock"
Repository = "https://github.com/simonlin1212/tradingagents-astock"
+423
View File
@@ -0,0 +1,423 @@
"""Tests for the claude_agent_sdk provider (personal Max-subscription POC).
The real Agent SDK spawns the `claude` CLI and consumes a live subscription, so
every test mocks the call path (`ClaudeAgentSDKClient._query`) — no CLI, no
network, no subscription needed.
conftest.py injects ANTHROPIC_API_KEY=placeholder for all tests; the F-004
guardrail trips on it, so tests that enable the provider must delenv it first
(this is the exact interaction Codex flagged).
"""
import json
import pytest
from pydantic import BaseModel
from tradingagents.llm_clients import claude_agent_sdk_client as mod
from tradingagents.llm_clients.claude_agent_sdk_client import (
AgentSDKChatModel,
ClaudeAgentSDKClient,
_RateLimitHit,
_split_prompt,
)
from tradingagents.llm_clients.factory import create_llm_client
class _Plan(BaseModel):
decision: str
confidence: int
@pytest.fixture
def oauth_env(monkeypatch):
"""Enable the provider cleanly: OAuth token present, API key absent."""
monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "test-oauth-token")
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
def _client_with_query(monkeypatch, text="", structured=None, fallback_spec=None):
"""Build a client whose _query returns a canned (text, structured) tuple."""
client = ClaudeAgentSDKClient("claude-opus-4-8", fallback_spec=fallback_spec)
async def fake_query(prompt, options, prefer_result=False):
return text, structured
monkeypatch.setattr(client, "_query", fake_query)
return client
class _FakeLangChainTool:
"""Minimal stand-in for a LangChain StructuredTool (bridged by bind_tools)."""
name = "get_thing"
description = "gets a thing"
args_schema = None # → _sdk_tools_from_langchain uses an empty object schema
def invoke(self, args):
return "thing-data"
# --------------------------------------------------------------------------- #
# T-002 / F-002: adapter surface
# --------------------------------------------------------------------------- #
def test_invoke_returns_aimessage(monkeypatch, oauth_env):
client = _client_with_query(monkeypatch, text="hello from max")
llm = client.get_llm()
result = llm.invoke("say hi")
assert result.content == "hello from max"
def test_structured_output_returns_pydantic(monkeypatch, oauth_env):
client = _client_with_query(
monkeypatch, structured={"decision": "buy", "confidence": 4}
)
llm = client.get_llm()
plan = llm.with_structured_output(_Plan).invoke("decide")
assert isinstance(plan, _Plan)
assert plan.decision == "buy" and plan.confidence == 4
def test_structured_output_parses_json_text_when_no_structured_field(monkeypatch, oauth_env):
# SDK returned text (not structured_output) — adapter must parse the JSON.
text = 'noise before {"decision": "hold", "confidence": 2} noise after'
client = _client_with_query(monkeypatch, text=text, structured=None)
plan = client.get_llm().with_structured_output(_Plan).invoke("decide")
assert plan.decision == "hold" and plan.confidence == 2
def test_bind_tools_returns_runnable_and_final_report(monkeypatch, oauth_env):
# bind_tools must return a Runnable (so `prompt | bound` composes) whose
# invoke runs the SDK tool loop and returns a final report with NO
# tool_calls — LangGraph then treats the analyst as done.
from langchain_core.runnables import Runnable
client = _client_with_query(monkeypatch, text="final report")
bound = client.get_llm().bind_tools([_FakeLangChainTool()])
assert isinstance(bound, Runnable)
result = bound.invoke("analyze 600519")
assert result.content == "final report"
assert result.tool_calls == []
def test_bind_tools_falls_back_on_rate_limit(monkeypatch, oauth_env):
# Subscription tool loop hits quota → fall back to the fallback provider's
# bind_tools, which rejoins LangGraph's normal external ToolNode loop.
client = ClaudeAgentSDKClient(
"claude-opus-4-8",
fallback_spec={"provider": "deepseek", "model": "deepseek-v4-pro", "base_url": None},
)
async def boom(prompt, options, prefer_result=False):
raise _RateLimitHit("weekly limit reached")
monkeypatch.setattr(client, "_query", boom)
_install_stub_fallback(monkeypatch)
result = client.get_llm().bind_tools([_FakeLangChainTool()]).invoke("analyze")
assert result.content == "served by fallback tools"
# --------------------------------------------------------------------------- #
# F-007 / AC-005: dependency + OAuth guards
# --------------------------------------------------------------------------- #
def test_get_llm_raises_when_sdk_missing(monkeypatch, oauth_env):
monkeypatch.setattr(mod, "_sdk", None)
client = ClaudeAgentSDKClient("claude-opus-4-8")
with pytest.raises(ImportError, match=r"\[agentsdk\]"):
client.get_llm()
def test_get_llm_uses_ambient_login_when_oauth_missing(monkeypatch):
# With no explicit token, get_llm() no longer raises — the Agent SDK inherits
# the ambient logged-in `claude` session (Keychain / ~/.claude). Any genuine
# auth failure surfaces at call time and triggers the F-005 fallback instead.
monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False)
# Only meaningful when the SDK is importable; otherwise the import guard wins.
if mod._sdk is None:
pytest.skip("claude-agent-sdk not installed")
client = ClaudeAgentSDKClient("claude-opus-4-8")
assert isinstance(client.get_llm(), mod.AgentSDKChatModel)
# --------------------------------------------------------------------------- #
# T-004: factory routing
# --------------------------------------------------------------------------- #
def test_factory_routes_to_agent_sdk_client():
client = create_llm_client(
"claude_agent_sdk", "claude-opus-4-8",
fallback_spec={"provider": "deepseek", "model": "deepseek-v4-pro", "base_url": None},
)
assert isinstance(client, ClaudeAgentSDKClient)
assert client.fallback_spec["provider"] == "deepseek"
# --------------------------------------------------------------------------- #
# T-006 / F-005 / AC-003: cross-provider fallback
# --------------------------------------------------------------------------- #
class _StubStructured:
def __init__(self, schema):
self._schema = schema
def invoke(self, prompt, *a, **k):
return self._schema(decision="fallback-buy", confidence=1)
class _StubBoundTools:
def invoke(self, prompt, *a, **k):
from langchain_core.messages import AIMessage
return AIMessage(content="served by fallback tools")
class _StubLLM:
def invoke(self, prompt, *a, **k):
from langchain_core.messages import AIMessage
return AIMessage(content="served by fallback")
def with_structured_output(self, schema, **k):
return _StubStructured(schema)
def bind_tools(self, tools, **k):
return _StubBoundTools()
def _install_stub_fallback(monkeypatch):
monkeypatch.setattr(
"tradingagents.llm_clients.factory.create_llm_client",
lambda **kw: type("C", (), {"get_llm": lambda self: _StubLLM()})(),
)
def test_invoke_falls_back_on_rate_limit(monkeypatch, oauth_env):
client = ClaudeAgentSDKClient(
"claude-opus-4-8",
fallback_spec={"provider": "deepseek", "model": "deepseek-v4-pro", "base_url": None},
)
async def boom(prompt, options):
raise _RateLimitHit("weekly limit reached")
monkeypatch.setattr(client, "_query", boom)
_install_stub_fallback(monkeypatch)
result = client.get_llm().invoke("analyze")
assert result.content == "served by fallback"
def test_structured_falls_back_and_still_yields_pydantic(monkeypatch, oauth_env):
client = ClaudeAgentSDKClient(
"claude-opus-4-8",
fallback_spec={"provider": "deepseek", "model": "deepseek-v4-pro", "base_url": None},
)
async def boom(prompt, options):
raise _RateLimitHit("weekly limit reached")
monkeypatch.setattr(client, "_query", boom)
_install_stub_fallback(monkeypatch)
plan = client.get_llm().with_structured_output(_Plan).invoke("decide")
assert isinstance(plan, _Plan)
assert plan.decision == "fallback-buy"
def test_no_fallback_spec_reraises(monkeypatch, oauth_env):
client = ClaudeAgentSDKClient("claude-opus-4-8", fallback_spec=None)
async def boom(prompt, options):
raise _RateLimitHit("limit")
monkeypatch.setattr(client, "_query", boom)
with pytest.raises(_RateLimitHit):
client.get_llm().invoke("x")
# --------------------------------------------------------------------------- #
# helpers
# --------------------------------------------------------------------------- #
def test_split_prompt_string():
assert _split_prompt("just text") == (None, "just text")
def test_split_prompt_messages_extracts_system():
from langchain_core.messages import SystemMessage, HumanMessage
system, user = _split_prompt([SystemMessage(content="be terse"), HumanMessage(content="hi")])
assert system == "be terse"
assert user == "hi"
# --------------------------------------------------------------------------- #
# T-005 / F-004: startup guardrail + AC-004 no-behaviour-change
# --------------------------------------------------------------------------- #
def test_guardrail_refuses_api_key_coexistence(monkeypatch):
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-would-bill-api")
monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "test-oauth-token")
from tradingagents.graph.trading_graph import TradingAgentsGraph
from tradingagents.default_config import DEFAULT_CONFIG
cfg = dict(DEFAULT_CONFIG)
cfg["deep_think_provider_override"] = "claude_agent_sdk"
with pytest.raises(RuntimeError, match="ANTHROPIC_API_KEY"):
TradingAgentsGraph(config=cfg)
def test_default_config_off_by_default():
from tradingagents.default_config import DEFAULT_CONFIG
assert DEFAULT_CONFIG["deep_think_provider_override"] is None
assert DEFAULT_CONFIG["quick_think_provider_override"] is None
# 默认用 claude CLI 的别名(opus/sonnet)而不是写死版本号的完整 id:
# 别名恒指向最新模型,写死 "claude-opus-4-8" 这类 id 会随版本迭代过期
# (仓库 model_catalog 里的 anthropic 条目就已经停在 4-6)。
# 完整 id 仍然可用,此处只锁默认值必须是别名。
assert DEFAULT_CONFIG["agent_sdk_model"] in ("opus", "sonnet")
# quick 节点数量多(7 分析师 + 辩手),订阅按额度限流,默认给更省的 sonnet。
assert DEFAULT_CONFIG["agent_sdk_quick_model"] == "sonnet"
# --------------------------------------------------------------------------- #
# real _query message consumption (closes the RateLimitEvent blind spot;
# locks in the fix for the "allowed_warning silently bills paid API" bug)
# --------------------------------------------------------------------------- #
def _fake(cls, **attrs):
"""Build an instance of a real SDK dataclass without its heavy __init__."""
obj = object.__new__(cls)
for k, v in attrs.items():
setattr(obj, k, v)
return obj
def _patch_query(monkeypatch, *messages):
async def fake_query(prompt, options):
for m in messages:
yield m
monkeypatch.setattr(mod._sdk, "query", fake_query)
def _rate_limit_info(status, overage_status=None):
# Set every field so repr()/str() (used in the reject path) never KeyErrors.
return _fake(
mod._sdk.RateLimitInfo,
status=status, resets_at=None, rate_limit_type=None, utilization=None,
overage_status=overage_status, overage_resets_at=None,
overage_disabled_reason=None, raw={},
)
@pytest.mark.skipif(mod._sdk is None, reason="claude-agent-sdk not installed")
def test_query_allowed_warning_does_not_fall_back(monkeypatch, oauth_env):
# Near the limit but STILL SERVING → must keep the response, not bail to API.
_patch_query(
monkeypatch,
_fake(mod._sdk.RateLimitEvent,
rate_limit_info=_rate_limit_info("allowed_warning")),
_fake(mod._sdk.AssistantMessage, content=[_fake(mod._sdk.TextBlock, text="served by subscription")]),
_fake(mod._sdk.ResultMessage, is_error=False, structured_output=None,
result="served by subscription", stop_reason="end_turn", api_error_status=None),
)
client = ClaudeAgentSDKClient(
"claude-opus-4-8",
fallback_spec={"provider": "deepseek", "model": "deepseek-v4-pro", "base_url": None},
)
_install_stub_fallback(monkeypatch)
result = client.get_llm().invoke("analyze")
assert result.content == "served by subscription" # NOT "served by fallback"
@pytest.mark.skipif(mod._sdk is None, reason="claude-agent-sdk not installed")
def test_query_allowed_with_overage_rejected_does_not_fall_back(monkeypatch, oauth_env):
# Real-world case: the org disables overage, so every event carries
# overage_status="rejected" — but status="allowed" means the plan served
# this call. Must keep the subscription response, NOT bail to the paid
# fallback. (Regression for the "100% silent downgrade" bug.)
_patch_query(
monkeypatch,
_fake(mod._sdk.RateLimitEvent,
rate_limit_info=_rate_limit_info("allowed", overage_status="rejected")),
_fake(mod._sdk.AssistantMessage, content=[_fake(mod._sdk.TextBlock, text="served by subscription")]),
_fake(mod._sdk.ResultMessage, is_error=False, structured_output=None,
result="served by subscription", stop_reason="end_turn", api_error_status=None),
)
client = ClaudeAgentSDKClient(
"claude-opus-4-8",
fallback_spec={"provider": "deepseek", "model": "deepseek-v4-pro", "base_url": None},
)
_install_stub_fallback(monkeypatch)
result = client.get_llm().invoke("analyze")
assert result.content == "served by subscription" # NOT "served by fallback"
@pytest.mark.skipif(mod._sdk is None, reason="claude-agent-sdk not installed")
def test_query_rejected_triggers_fallback(monkeypatch, oauth_env):
_patch_query(
monkeypatch,
_fake(mod._sdk.RateLimitEvent,
rate_limit_info=_rate_limit_info("rejected", overage_status="rejected")),
)
client = ClaudeAgentSDKClient(
"claude-opus-4-8",
fallback_spec={"provider": "deepseek", "model": "deepseek-v4-pro", "base_url": None},
)
_install_stub_fallback(monkeypatch)
result = client.get_llm().invoke("analyze")
assert result.content == "served by fallback"
# --------------------------------------------------------------------------- #
# 认证失败必须中止,不能降级到按 token 计费的 provider
# --------------------------------------------------------------------------- #
def test_auth_failure_is_detected_from_api_retry_and_assistant_text():
from tradingagents.llm_clients.claude_agent_sdk_client import (
_looks_like_auth_failure,
)
class _Sys: # SystemMessage(subtype="api_retry")
data = {"error_status": 401, "error": "authentication_failed"}
class _Block:
text = "Failed to authenticate. API Error: 401 OAuth access token has expired."
class _Assistant:
content = [_Block()]
error = "authentication_error"
class _Normal:
data = {"subtype": "init"}
content = []
error = None
assert _looks_like_auth_failure(_Sys()) is True
assert _looks_like_auth_failure(_Assistant()) is True
assert _looks_like_auth_failure(_Normal()) is False
def test_auth_error_is_not_in_fallback_errors():
"""凭据失效时降级到付费 provider = 悄悄开始计费,正是启用订阅要避免的。
这条锁住 _AuthError 不被 fallback 吞掉。"""
from tradingagents.llm_clients import claude_agent_sdk_client as mod
assert mod._AuthError not in mod._FALLBACK_ERRORS
assert not issubclass(mod._AuthError, mod._FALLBACK_ERRORS)
def test_auth_failure_hint_is_actionable():
from tradingagents.llm_clients.claude_agent_sdk_client import _auth_failure_hint
class _Block:
text = "401 OAuth access token has expired"
class _M:
content = [_Block()]
hint = _auth_failure_hint(_M())
assert "claude setup-token" in hint # 给出可执行命令
assert "CLAUDE_CODE_OAUTH_TOKEN" in hint
+19
View File
@@ -25,6 +25,25 @@ DEFAULT_CONFIG = {
"google_thinking_level": None, # "high", "minimal", etc.
"openai_reasoning_effort": None, # "medium", "high", "low"
"anthropic_effort": None, # "high", "medium", "low"
# ── Claude Agent SDK provider(走个人 Pro/Max 订阅额度,可选依赖 [agentsdk])──
# 与内置 anthropic provider 的区别:anthropic 走 ANTHROPIC_API_KEY = **按 token 计费**;
# 本 provider 走本机已登录的 claude CLI = **消耗订阅额度,不产生 API 账单**。
# 设为 "claude_agent_sdk" 时,仅 deep_thinking_llm 节点(Research Manager /
# Portfolio Manager)走订阅。None = 维持原行为。
"deep_think_provider_override": None,
# 同上,作用于 QUICK 节点(7 个工具分析师 + 多空/交易员/风险辩手)。
# 与上一项同时开启 = 全节点走订阅。None = 分析师仍走 llm_provider(维持原行为)。
"quick_think_provider_override": None,
# Agent SDK 使用的 Claude 模型。**必须是真实 Claude 模型**,不要复用 deep_think_llm。
# 用别名而非写死版本号:claude CLI 的 "opus"/"sonnet" 恒指向最新模型,
# 写死 "claude-opus-4-8" 这类 ID 会随版本迭代过期。
"agent_sdk_model": "opus",
# QUICK/分析师节点用的 Claude 模型。默认 sonnet 而非 opus——quick 节点数量多
# (7 分析师 + 辩手),订阅是按额度限流的,全用 opus 很快会撞到上限。
"agent_sdk_quick_model": "sonnet",
# 订阅调用失败 / 撞额度时的兜底。None → 回落到 llm_provider + deep_think_llm。
"agent_sdk_fallback_provider": None,
"agent_sdk_fallback_model": None,
# Checkpoint/resume: when True, LangGraph saves state after each node
# so a crashed run can resume from the last successful step.
"checkpoint_enabled": False,
+44 -12
View File
@@ -163,18 +163,50 @@ class TradingAgentsGraph:
if self.callbacks:
llm_kwargs["callbacks"] = self.callbacks
deep_client = create_llm_client(
provider=self.config["llm_provider"],
model=self.config["deep_think_llm"],
base_url=self.config.get("backend_url"),
**llm_kwargs,
)
quick_client = create_llm_client(
provider=self.config["llm_provider"],
model=self.config["quick_think_llm"],
base_url=self.config.get("backend_url"),
**llm_kwargs,
)
# Optional: route nodes through a personal Claude Pro/Max subscription
# via the Claude Agent SDK. `deep_think_provider_override` covers the
# deep nodes (Research / Portfolio Manager); `quick_think_provider_override`
# covers the quick nodes (7 tool-using analysts + Bull/Bear / trader /
# risk debaters). Both on ⇒ every node runs on the subscription. Off by
# default — behaviour is unchanged when both are None.
deep_on = self.config.get("deep_think_provider_override") == "claude_agent_sdk"
quick_on = self.config.get("quick_think_provider_override") == "claude_agent_sdk"
# F-004 guardrail: ANTHROPIC_API_KEY outranks the subscription OAuth token
# and would silently bill the pay-per-token API instead of the subscription.
# Refuse to start rather than surprise-bill the user.
if (deep_on or quick_on) and os.getenv("ANTHROPIC_API_KEY"):
raise RuntimeError(
"claude_agent_sdk override is on but ANTHROPIC_API_KEY is set — the "
"API key takes precedence and would silently bill the Anthropic API "
"instead of your Pro/Max subscription. Unset ANTHROPIC_API_KEY to use "
"the subscription, or clear the override."
)
def _make_client(override_on, sdk_model_key, fallback_model_key):
"""Build a subscription-backed client when overridden, else the normal
llm_provider client. Fallback rejoins the paid provider on quota/failure."""
if override_on:
fallback_spec = {
"provider": self.config.get("agent_sdk_fallback_provider") or self.config["llm_provider"],
"model": self.config.get("agent_sdk_fallback_model") or self.config[fallback_model_key],
"base_url": self.config.get("backend_url"),
}
return create_llm_client(
provider="claude_agent_sdk",
model=self.config[sdk_model_key],
base_url=self.config.get("backend_url"),
fallback_spec=fallback_spec,
)
return create_llm_client(
provider=self.config["llm_provider"],
model=self.config[fallback_model_key],
base_url=self.config.get("backend_url"),
**llm_kwargs,
)
deep_client = _make_client(deep_on, "agent_sdk_model", "deep_think_llm")
quick_client = _make_client(quick_on, "agent_sdk_quick_model", "quick_think_llm")
self.deep_thinking_llm = deep_client.get_llm()
self.quick_thinking_llm = quick_client.get_llm()
@@ -0,0 +1,533 @@
"""Claude Agent SDK provider — route deep-thinking nodes through a personal
Claude Pro/Max subscription instead of the pay-per-token Anthropic API.
Personal use only. The Claude Agent SDK / ``claude -p`` consume the logged-in
user's subscription quota (obtained via ``claude setup-token`` →
``CLAUDE_CODE_OAUTH_TOKEN``). Publishing this as a product that routes *other*
users' subscription credentials requires Anthropic approval — out of scope.
Packaged as the optional ``[agentsdk]`` extra: it pulls a CLI-backed
dependency chain (claude-agent-sdk → mcp → httpx2) that most users do not
need. Verified 2026-07-31 that this does **not** clash with mootdx's
``httpx<0.26`` pin — mcp moved to ``httpx2`` — so unlike the ``[google]``
extra removed in #87, ``uv lock`` resolves cleanly with it declared.
The import below is guarded so the base install never breaks; the error
surfaces only when this provider is actually selected.
Scope: originally the ``deep_thinking_llm`` nodes (Research Manager / Portfolio
Manager) which call ``.invoke(str)`` / ``.with_structured_output`` directly.
``bind_tools`` is now also supported for the tool-using analysts: their
LangChain tools are bridged to Agent SDK in-process MCP tools and the SDK runs
the ReAct loop internally, returning a final report (no tool_calls) so
LangGraph treats the analyst as done. On failure/quota the fallback provider's
``bind_tools`` rejoins LangGraph's normal external ToolNode loop.
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import re
import threading
from typing import Any, Optional
from langchain_core.messages import AIMessage
from langchain_core.runnables import Runnable
from .base_client import BaseLLMClient
logger = logging.getLogger(__name__)
OAUTH_ENV = "CLAUDE_CODE_OAUTH_TOKEN"
# Tool-loop knobs. When the deep-thinking nodes call plain .invoke there are no
# tools and a single turn suffices; the tool-using analysts need the Agent SDK
# to run a multi-turn agentic loop internally (call tool → read result → …).
_MCP_SERVER_NAME = "astock_tools"
_TOOL_MAX_TURNS = 30 # generous: an analyst may pull ~8 indicators + data
_TOOL_RESULT_CAP = 60_000 # per-tool result char cap (safety, not normally hit)
try: # optional dependency — see module docstring
import claude_agent_sdk as _sdk
from claude_agent_sdk import (
ClaudeAgentOptions,
ClaudeSDKError,
create_sdk_mcp_server,
tool as _sdk_tool,
)
_IMPORT_ERROR: Optional[Exception] = None
except Exception as exc: # ImportError or any transitive import failure
_sdk = None
ClaudeAgentOptions = None
create_sdk_mcp_server = None
_sdk_tool = None
ClaudeSDKError = Exception # placeholder so `except` clauses never NameError
_IMPORT_ERROR = exc
_AUTH_MARKERS = (
"authentication_failed",
"oauth access token has expired",
"re-authenticate",
"invalid api key",
"please run /login",
)
def _looks_like_auth_failure(message: Any) -> bool:
"""识别订阅凭据失效。覆盖两条路径:
① SystemMessage(subtype="api_retry") 带 error="authentication_failed" / status 401
② AssistantMessage 的文本里出现 401 / 需重新认证的措辞(model="<synthetic>")
"""
data = getattr(message, "data", None)
if isinstance(data, dict):
if data.get("error_status") == 401 or data.get("error") == "authentication_failed":
return True
blob = ""
content = getattr(message, "content", None)
if isinstance(content, list):
blob = " ".join(
getattr(b, "text", "") for b in content if getattr(b, "text", "")
)
if getattr(message, "error", None):
blob += f" {message.error}"
blob = blob.lower()
return any(m in blob for m in _AUTH_MARKERS)
def _auth_failure_hint(message: Any) -> str:
detail = ""
content = getattr(message, "content", None)
if isinstance(content, list):
detail = " ".join(
getattr(b, "text", "") for b in content if getattr(b, "text", "")
).strip()
if not detail:
data = getattr(message, "data", None)
if isinstance(data, dict):
detail = str(data.get("error") or data.get("error_status") or "")
return (
"Claude 订阅凭据失效,无法走订阅额度"
+ (f"({detail})" if detail else "")
+ "。已中止而**不是**降级到按 token 计费的 provider——"
"你启用订阅模式就是为了避免 API 账单,静默降级等于悄悄开始计费。\n"
"修复:在终端跑 `claude setup-token`(需已登录 Pro/Max 账号),"
"把输出的 token 设为 CLAUDE_CODE_OAUTH_TOKEN;或直接 `claude` 重新登录。\n"
"若确实想用按量计费的 Anthropic API,请把 provider 改回 anthropic。"
)
class _RateLimitHit(Exception):
"""Raised when the subscription hit a rate/quota limit — triggers fallback."""
class _SDKResultError(Exception):
"""Raised when the Agent SDK returns an error ResultMessage — triggers fallback."""
# Errors that mean "the subscription path could not serve this call" → fall back.
class _AuthError(Exception):
"""订阅凭据失效(OAuth token 过期 / 未登录)。
⚠️ **刻意不放进 _FALLBACK_ERRORS**:用户开订阅模式就是为了不产生 API 账单,
token 一过期就静默降级到按 token 计费的 provider = 悄悄开始烧钱,
正是启动护栏 F-004 想防的事,只是从启动时挪到了运行中。
这里直接中止并告诉用户怎么重新登录。
"""
# 认证失败**不在**此列表:只有限流/SDK 故障才降级到付费 provider。
_FALLBACK_ERRORS = (ClaudeSDKError, _RateLimitHit, _SDKResultError)
# --------------------------------------------------------------------------- #
# prompt/message helpers
# --------------------------------------------------------------------------- #
def _msg_role_content(message: Any):
"""Extract (role, content) from a LangChain BaseMessage or a plain dict."""
if isinstance(message, dict):
return message.get("role"), str(message.get("content", ""))
role = getattr(message, "type", None) # BaseMessage.type: 'system'/'human'/'ai'
content = getattr(message, "content", "")
return role, content if isinstance(content, str) else str(content)
def _split_prompt(prompt: Any):
"""Return (system_prompt_or_None, user_text) from whatever the agents pass.
The deep-thinking agents pass a plain string, but accept PromptValue and
message lists defensively so this never crashes on an unexpected shape.
"""
if isinstance(prompt, str):
return None, prompt
to_messages = getattr(prompt, "to_messages", None)
if callable(to_messages):
messages = to_messages()
elif isinstance(prompt, (list, tuple)):
messages = list(prompt)
else:
return None, str(prompt)
system_parts, other_parts = [], []
for m in messages:
role, content = _msg_role_content(m)
if role == "system":
system_parts.append(content)
elif role in (None, "human", "user", "ai", "assistant"):
other_parts.append(content)
else:
other_parts.append(f"[{role}] {content}")
system = "\n".join(p for p in system_parts if p) or None
return system, "\n".join(p for p in other_parts if p)
_JSON_RE = re.compile(r"\{.*\}", re.DOTALL)
def _extract_json(text: str) -> str:
"""Best-effort: pull the first {...} block out of a text response."""
match = _JSON_RE.search(text or "")
if not match:
raise ValueError("no JSON object found in Agent SDK text response")
return match.group(0)
def _run_async(coro):
"""Run an async coroutine to completion from a synchronous caller.
``trading_graph`` drives LangGraph synchronously, so normally there is no
running loop and ``asyncio.run`` works. If a loop is already running, run
the coroutine on a dedicated thread with its own loop so we never disturb
the caller's loop.
"""
try:
asyncio.get_running_loop()
except RuntimeError:
return asyncio.run(coro)
box: dict[str, Any] = {}
def _worker():
try:
box["value"] = asyncio.run(coro)
except BaseException as exc: # propagate to the calling thread, don't swallow
box["error"] = exc
thread = threading.Thread(target=_worker)
thread.start()
thread.join()
if "error" in box:
raise box["error"]
return box["value"]
def _sdk_tools_from_langchain(lc_tools):
"""Wrap LangChain tools as Agent SDK in-process MCP tools.
The analysts drive tools via LangGraph's external ReAct loop (return
tool_calls → ToolNode executes → repeat). The Agent SDK instead runs the
whole loop *internally*, so we register each LangChain tool as an SDK tool
whose async handler invokes the original tool on a worker thread (the data
layer is synchronous and hits the network — never block the SDK's loop).
"""
sdk_tools = []
for lc in lc_tools:
schema = (
lc.args_schema.model_json_schema()
if getattr(lc, "args_schema", None) is not None
else {"type": "object", "properties": {}}
)
@_sdk_tool(lc.name, (lc.description or lc.name)[:1000], schema)
async def _handler(args, _lc=lc): # _lc default-arg pins the loop var
try:
result = await asyncio.to_thread(_lc.invoke, dict(args))
except Exception as exc: # tool failure is data for the model, not a crash
result = f"[tool error] {exc}"
return {"content": [{"type": "text", "text": str(result)[:_TOOL_RESULT_CAP]}]}
sdk_tools.append(_handler)
return sdk_tools
# --------------------------------------------------------------------------- #
# adapters returned to the agents
# --------------------------------------------------------------------------- #
class _StructuredAgentSDK:
"""What ``with_structured_output(schema)`` returns — exposes ``.invoke``."""
def __init__(self, adapter: "AgentSDKChatModel", schema):
self._adapter = adapter
self._schema = schema
def invoke(self, prompt: Any, *args, **kwargs):
try:
return self._adapter._client._invoke_structured(self._schema, prompt)
except _FALLBACK_ERRORS as exc:
fallback = self._adapter._get_fallback()
if fallback is None:
raise
logger.warning(
"claude_agent_sdk: structured call failed (%s); "
"falling back to provider '%s'",
exc, self._adapter._fallback_desc(),
)
return fallback.with_structured_output(self._schema).invoke(prompt, *args, **kwargs)
class _BoundAgentSDK(Runnable):
"""What ``bind_tools(tools)`` returns — a Runnable so ``prompt | bound`` works.
``.invoke`` runs the Agent SDK's internal tool loop on the subscription and
returns the final report as an ``AIMessage`` with no ``tool_calls``. On a
subscription failure/quota it defers to the fallback provider, whose
``bind_tools`` returns real ``tool_calls`` and so rejoins LangGraph's normal
external ToolNode loop — no graph change needed either way.
"""
def __init__(self, adapter: "AgentSDKChatModel", tools):
self._adapter = adapter
self._tools = tools
def invoke(self, input, config=None, **kwargs):
try:
return self._adapter._client._invoke_with_tools(self._tools, input)
except _FALLBACK_ERRORS as exc:
fallback = self._adapter._get_fallback()
if fallback is None:
raise
logger.warning(
"claude_agent_sdk: tool invoke failed (%s); "
"falling back to provider '%s'",
exc, self._adapter._fallback_desc(),
)
return fallback.bind_tools(self._tools).invoke(input, config, **kwargs)
class AgentSDKChatModel:
"""Duck-typed LangChain-compatible chat model backed by the Claude Agent SDK.
Only the surface the deep-thinking agents use is implemented: ``invoke``,
``with_structured_output`` and ``bind_tools`` (the last raises on purpose).
Cross-provider fallback (F-005) is self-contained here so callers never see
a subscription quota error.
"""
def __init__(self, client: "ClaudeAgentSDKClient", fallback_spec: Optional[dict]):
self._client = client
self._fallback_spec = fallback_spec or None
self._fallback_llm = None # built lazily on first failure
def _fallback_desc(self) -> str:
return (self._fallback_spec or {}).get("provider", "none")
def _get_fallback(self):
if self._fallback_spec is None:
return None
if self._fallback_llm is None:
from .factory import create_llm_client
self._fallback_llm = create_llm_client(**self._fallback_spec).get_llm()
return self._fallback_llm
def invoke(self, prompt: Any, *args, **kwargs):
try:
return self._client._invoke_raw(prompt)
except _FALLBACK_ERRORS as exc:
fallback = self._get_fallback()
if fallback is None:
raise
logger.warning(
"claude_agent_sdk: invoke failed (%s); falling back to provider '%s'",
exc, self._fallback_desc(),
)
return fallback.invoke(prompt, *args, **kwargs)
def with_structured_output(self, schema, **kwargs):
return _StructuredAgentSDK(self, schema)
def bind_tools(self, tools, **kwargs):
# The tool-using analysts compose `prompt | llm.bind_tools(tools)`, so
# this must return a Runnable. The Agent SDK runs the tool loop
# internally and returns a final report (no tool_calls) — LangGraph
# then treats the analyst as done. On failure/quota, the fallback
# provider's bind_tools rejoins the normal external ToolNode loop.
return _BoundAgentSDK(self, list(tools))
# --------------------------------------------------------------------------- #
# client
# --------------------------------------------------------------------------- #
class ClaudeAgentSDKClient(BaseLLMClient):
"""LLM client that calls Claude via the Agent SDK on a personal subscription.
``fallback_spec`` (``{"provider", "model", "base_url"}``) is injected at
construction so the adapter can rebuild a fallback LLM internally without
reaching back into ``trading_graph``.
"""
def __init__(self, model: str, base_url: Optional[str] = None,
fallback_spec: Optional[dict] = None, **kwargs):
super().__init__(model, base_url, **kwargs)
self.fallback_spec = fallback_spec
def get_llm(self) -> AgentSDKChatModel:
if _sdk is None:
raise ImportError(
"claude-agent-sdk is not installed. Install the optional extra:\n"
' pip install -e ".[agentsdk]"\n'
f"(original import error: {_IMPORT_ERROR})"
)
if not os.getenv(OAUTH_ENV):
# No explicit subscription token — the Agent SDK spawns the `claude`
# CLI, which inherits the ambient logged-in session (macOS Keychain /
# ~/.claude credentials). That is the SAME personal subscription path;
# `claude setup-token` merely pins an explicit, portable token needed
# on headless/CI boxes with no logged-in CLI. Proceed and let the SDK
# surface an auth error at call time (F-005 fallback catches it)
# rather than block a machine that is already logged in.
logger.info(
"%s not set — using the ambient logged-in `claude` session "
"(personal subscription). Run `claude setup-token` to pin an "
"explicit token for headless use.",
OAUTH_ENV,
)
return AgentSDKChatModel(self, self.fallback_spec)
def validate_model(self) -> bool:
# Any Claude model id is accepted; validity is enforced by the SDK/CLI.
return True
# -- internal call path -------------------------------------------------- #
def _build_options(self, system_prompt: Optional[str],
output_format: Optional[dict] = None,
sdk_tools: Optional[list] = None,
tool_names: Optional[list] = None):
opts: dict[str, Any] = {
"model": self.model,
"max_turns": 1, # approximate a single completion
"allowed_tools": [], # no built-in tools (Read/Write/Bash/…)
"setting_sources": [], # don't load filesystem settings/skills/CLAUDE.md
"permission_mode": "bypassPermissions",
}
token = os.environ.get(OAUTH_ENV)
if token:
# Pin the explicit subscription token when provided; otherwise let the
# SDK fall back to the ambient logged-in CLI session (Keychain) —
# passing an empty token here is unnecessary and only muddies intent.
opts["env"] = {OAUTH_ENV: token}
if sdk_tools:
# Register the bridged analyst tools and let the SDK run the loop
# internally over multiple turns (only these tools are allowed).
server = create_sdk_mcp_server(_MCP_SERVER_NAME, "1.0.0", tools=sdk_tools)
opts["mcp_servers"] = {_MCP_SERVER_NAME: server}
opts["allowed_tools"] = [
f"mcp__{_MCP_SERVER_NAME}__{n}" for n in (tool_names or [])
]
opts["max_turns"] = _TOOL_MAX_TURNS
if system_prompt:
opts["system_prompt"] = system_prompt
if output_format is not None:
opts["output_format"] = output_format
return ClaudeAgentOptions(**opts)
async def _query(self, prompt: str, options, prefer_result: bool = False):
text_parts: list[str] = []
result_msg = None
auth_hint: Optional[str] = None
async for message in _sdk.query(prompt=prompt, options=options):
if isinstance(message, _sdk.RateLimitEvent):
# RateLimitEvent fires on ANY status change, including
# "allowed_warning" (near the limit but STILL SERVING) and
# "allowed" (recovered). Only status=="rejected" means THIS call
# was actually blocked. Raising on anything else discards a
# successful subscription response and silently falls back to
# the paid provider — the opposite of this feature's purpose.
#
# In particular, DO NOT key off overage_status: it is a separate
# axis describing whether *overage* (paid usage beyond the plan)
# is available. When an org disables overage it reports
# overage_status="rejected"/"org_level_disabled" on EVERY event,
# including status=="allowed" ones the plan served within
# allowance — so keying fallback off it downgraded 100% of
# subscription calls to the paid fallback. (regression:
# test_query_allowed_with_overage_rejected_does_not_fall_back)
info = getattr(message, "rate_limit_info", None)
if getattr(info, "status", None) == "rejected":
raise _RateLimitHit(str(info))
logger.warning(
"claude_agent_sdk: rate-limit status=%s (still serving); continuing",
getattr(info, "status", None),
)
continue
# 认证失败在 SDK 里会被翻译成 "error result: success" 这种毫无信息量的
# 报错(实测 2026-07-31:OAuth token 过期时 ResultMessage.subtype 仍是
# "success"、is_error=True,真正的原因只出现在 api_retry 事件和助手文本里)。
# 在这里正向识别,给出可执行的修复指引。
if _looks_like_auth_failure(message):
# 先跳出再抛:在 async for 内部抛异常会让 SDK 的异步生成器
# 处于运行中被关闭的状态,附带一条 "aclose(): asynchronous
# generator is already running" 噪音,掩盖真正的原因。
auth_hint = _auth_failure_hint(message)
break
if isinstance(message, _sdk.AssistantMessage):
for block in message.content:
if isinstance(block, _sdk.TextBlock):
text_parts.append(block.text)
elif isinstance(message, _sdk.ResultMessage):
result_msg = message
if auth_hint:
raise _AuthError(auth_hint)
structured = None
text = "".join(text_parts)
if result_msg is not None:
if getattr(result_msg, "is_error", False):
raise _SDKResultError(
f"stop_reason={getattr(result_msg, 'stop_reason', None)} "
f"api_error_status={getattr(result_msg, 'api_error_status', None)}"
)
structured = getattr(result_msg, "structured_output", None)
final = getattr(result_msg, "result", None)
# In a tool loop the intermediate turns emit reasoning text before
# each tool call; ResultMessage.result holds the authoritative final
# answer, so prefer it there. For single-turn calls fall back to it
# only when no assistant text was streamed.
if final and (prefer_result or not text):
text = final
return text, structured
def _invoke_raw(self, prompt: Any) -> AIMessage:
system_prompt, user_text = _split_prompt(prompt)
options = self._build_options(system_prompt)
text, _ = _run_async(self._query(user_text, options))
return AIMessage(content=text)
def _invoke_with_tools(self, lc_tools, prompt: Any) -> AIMessage:
"""Run the SDK's internal tool loop over the bridged analyst tools and
return the final report as an AIMessage (no tool_calls)."""
system_prompt, user_text = _split_prompt(prompt)
sdk_tools = _sdk_tools_from_langchain(lc_tools)
tool_names = [t.name for t in lc_tools]
options = self._build_options(
system_prompt, sdk_tools=sdk_tools, tool_names=tool_names
)
text, _ = _run_async(self._query(user_text, options, prefer_result=True))
return AIMessage(content=text)
def _invoke_structured(self, schema, prompt: Any):
system_prompt, user_text = _split_prompt(prompt)
output_format = {"type": "json_schema", "schema": schema.model_json_schema()}
options = self._build_options(system_prompt, output_format=output_format)
text, structured = _run_async(self._query(user_text, options))
data = structured if structured is not None else json.loads(_extract_json(text))
return schema.model_validate(data)
+4
View File
@@ -47,6 +47,10 @@ def create_llm_client(
from .anthropic_client import AnthropicClient
return AnthropicClient(model, base_url, **kwargs)
if provider_lower == "claude_agent_sdk":
from .claude_agent_sdk_client import ClaudeAgentSDKClient
return ClaudeAgentSDKClient(model, base_url, **kwargs)
if provider_lower == "google":
from .google_client import GoogleClient
return GoogleClient(model, base_url, **kwargs)
+12
View File
@@ -177,6 +177,18 @@ def _build_config() -> dict:
config["max_risk_discuss_rounds"] = 1
config["checkpoint_enabled"] = True
config["output_language"] = "Chinese"
# Optional: route nodes through a personal Claude Pro/Max subscription (Agent
# SDK). Scope: "deep" = Research/Portfolio only; "all" = + the 7 analysts.
# Leaving the fallback keys None makes the graph fall back to the
# sidebar-selected llm_provider + models on quota/failure.
scope = st.session_state.get("subscription_scope", "off")
sub_model = st.session_state.get("agent_sdk_model") or "claude-opus-4-8"
if scope in ("deep", "all"):
config["deep_think_provider_override"] = "claude_agent_sdk"
config["agent_sdk_model"] = sub_model
if scope == "all":
config["quick_think_provider_override"] = "claude_agent_sdk"
config["agent_sdk_quick_model"] = sub_model
return config
+44
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import os
from datetime import date
import streamlit as st
@@ -197,6 +198,49 @@ def _render_llm_config() -> None:
"Completions),模型 ID 手动填写,Key 在 .env 设 `OPENAI_COMPATIBLE_API_KEY`。"
)
# ── 个人 Claude 订阅额度(可选,仅个人自用)────────────────────────
_scope_labels = [
"关闭(走上面选的供应商)",
"仅深度节点(Research/Portfolio)",
"所有节点(含 7 个工具分析师)",
]
_scope_values = ["off", "deep", "all"]
scope_idx = st.selectbox(
"个人 Claude 订阅覆盖 (Agent SDK)",
range(len(_scope_labels)),
format_func=lambda i: _scope_labels[i],
key="subscription_scope_idx",
help=(
"让部分/全部节点经 Claude Agent SDK 走你个人 Pro/Max 订阅额度,"
"而非按 token 计费。「所有节点」含 7 个工具分析师(其工具调用已桥接到订阅)。"
"需装 [agentsdk] 依赖,且本机 claude 已登录(或设 CLAUDE_CODE_OAUTH_TOKEN)。"
),
)
scope = _scope_values[scope_idx]
st.session_state["subscription_scope"] = scope
if scope != "off":
# 用别名而非写死版本号:claude CLI 的 opus/sonnet 恒指向最新模型。
st.session_state.setdefault("agent_sdk_model", "opus")
st.text_input(
"订阅使用的 Claude 模型",
key="agent_sdk_model",
help=(
"填别名 opus / sonnet(恒指向最新模型,推荐)或完整模型 id。"
"撞额度/失败时自动降级到上面选的供应商 + 对应模型。"
),
)
if scope == "all":
st.caption(
"⚠️ 「所有节点」会把 7 个分析师 + 多空/交易员/风险辩手全部压到订阅上,"
"订阅是按额度限流的,跑几轮就可能撞上限。可在 config 里把 "
"`agent_sdk_quick_model` 设为 `sonnet` 降低消耗(默认已是)。"
)
if os.getenv("ANTHROPIC_API_KEY"):
st.warning(
"检测到 ANTHROPIC_API_KEY —— 启用订阅时它会被优先、悄悄走 API 计费,"
"启动护栏会直接报错中止。请在 .env 清空 ANTHROPIC_API_KEY 后重启 Web。"
)
def render_sidebar() -> None:
"""Render the sidebar with input controls and history."""