feat(p8.4): agentic runner 启动模式 - 复用 mcp_tools + Claude Agent SDK

ZhaoXingPeng · ZhaoXingPeng · commit 34d1c0fb9d8b · 2026-06-03T21:32:00.000+08:00
新增 server/agentic_runner.py:
- AgenticConfig: goal / model / max_iterations / allowed_tools / dry_run
- AgenticResult: 成功标志 + 迭代数 + tool_calls + 错误信息
- is_agentic_available(): 探测 ANTHROPIC_API_KEY + claude-agent-sdk 是否就绪
- run_agentic(): 主入口,SDK 缺失时优雅退出而非异常
- 复用 database.mcp_tools 工具注册表,zero duplication
- dry_run 模式不调 LLM,只打印 plan (CI / 配置校验用)

启动模式对比:
- mcp_server.py: 经典 MCP server,被动等待客户端调用
- agentic_runner.py: 主动 agent,内嵌 SDK,自己规划/调用/迭代

13 个 unit test 覆盖配置校验 / 可用性探测 / dry_run / 工具过滤 / 不可用降级。
不依赖真实 ANTHROPIC_API_KEY 或 SDK 安装。
diff --git a/src/dbjavagenix/server/agentic_runner.py b/src/dbjavagenix/server/agentic_runner.py
@@ -0,0 +1,179 @@
+"""
+P8.4: Agentic runner mode using Claude Agent SDK.
+
+启动模式对比:
+- `mcp_server.py`: 经典 MCP server,被动等待客户端调用 tools
+- `agentic_runner.py`: 主动 agent,内嵌 Claude Agent SDK,自己规划/调用/迭代
+
+用途:
+- 一次性批处理:"分析 host=X 的 schema,生成代码到 ./out" 单条命令完成全流程
+- 离线/CI:无人值守生成 (需要 ANTHROPIC_API_KEY)
+- 教学 demo:让面试官看到"同样的 tools 既能被 Claude Desktop 用,
+  也能被独立 agent 用"
+
+设计:
+- 复用 mcp_tools 的所有工具定义 (zero duplication)
+- Agent SDK 不可用 (没装 claude-agent-sdk 或没 ANTHROPIC_API_KEY) 时
+  优雅退出并提示用户走 mcp_server 模式
+- 不直接 import claude_agent_sdk - 用 lazy import,避免没装 SDK 的开发环境
+  import 项目就报错
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from dataclasses import dataclass
+from typing import Any, Callable, Optional
+
+logger = logging.getLogger(__name__)
+
+
+# ============================================================
+# 可用性探测
+# ============================================================
+
+def is_agentic_available() -> tuple[bool, str]:
+    """返回 (是否可用, 不可用原因)。
+
+    可用条件:
+    1. ANTHROPIC_API_KEY 环境变量存在
+    2. claude_agent_sdk 包已安装
+    """
+    if not os.environ.get("ANTHROPIC_API_KEY"):
+        return False, "ANTHROPIC_API_KEY not set"
+    try:
+        import claude_agent_sdk  # noqa: F401
+        return True, ""
+    except ImportError:
+        return False, "claude-agent-sdk not installed (pip install claude-agent-sdk)"
+
+
+# ============================================================
+# 配置 + 结果
+# ============================================================
+
+@dataclass
+class AgenticConfig:
+    """Agentic runner 启动参数"""
+    goal: str  # 自然语言目标,例如 "分析 host=X 的 schema 并生成代码到 ./out"
+    model: str = "claude-opus-4-7"
+    max_iterations: int = 20
+    allowed_tools: Optional[list[str]] = None  # None = 所有 tools
+    dry_run: bool = False  # True 时不真正调 LLM,只打印 plan
+
+    def __post_init__(self) -> None:
+        if not self.goal or not isinstance(self.goal, str):
+            raise ValueError("goal must be non-empty string")
+        if self.max_iterations < 1 or self.max_iterations > 100:
+            raise ValueError("max_iterations must be in [1, 100]")
+
+
+@dataclass
+class AgenticResult:
+    """Agent 跑完后的总结"""
+    success: bool
+    iterations_used: int = 0
+    final_message: str = ""
+    tool_calls_made: int = 0
+    error: Optional[str] = None
+
+
+# ============================================================
+# 主入口
+# ============================================================
+
+def run_agentic(
+    config: AgenticConfig,
+    tool_registry_factory: Callable[[], dict[str, Any]] | None = None,
+) -> AgenticResult:
+    """启动 agentic runner。
+
+    Args:
+        config: 启动配置
+        tool_registry_factory: 返回 tool 注册表的工厂函数,默认用 dbjavagenix 自带
+            的全套 tools。测试时可注入空 registry。
+
+    Returns:
+        AgenticResult: 跑完(成功 / 达到 max_iterations / 出错)的总结
+    """
+    available, reason = is_agentic_available()
+    if not available:
+        return AgenticResult(
+            success=False,
+            error=f"agentic mode unavailable: {reason}",
+        )
+
+    if config.dry_run:
+        # dry_run 不调 SDK,只把 plan 打出来。便于 CI 验证配置正确性
+        registry = tool_registry_factory() if tool_registry_factory else {}
+        tool_count = len(config.allowed_tools or registry.keys())
+        return AgenticResult(
+            success=True,
+            iterations_used=0,
+            final_message=(
+                f"[dry-run] goal={config.goal!r} model={config.model} "
+                f"max_iter={config.max_iterations} tools={tool_count}"
+            ),
+            tool_calls_made=0,
+        )
+
+    # 真实启动 — lazy import 避免没装 SDK 时 import 此模块直接炸
+    try:
+        from claude_agent_sdk import Agent  # type: ignore
+    except ImportError as e:  # pragma: no cover - 被 is_agentic_available 拦住
+        return AgenticResult(success=False, error=f"import failed: {e}")
+
+    registry = tool_registry_factory() if tool_registry_factory else _default_tool_registry()
+    try:
+        agent = Agent(
+            model=config.model,
+            tools=_filter_tools(registry, config.allowed_tools),
+            max_iterations=config.max_iterations,
+        )
+        result = agent.run(config.goal)
+        return AgenticResult(
+            success=True,
+            iterations_used=getattr(result, "iterations", 0),
+            final_message=getattr(result, "final_message", ""),
+            tool_calls_made=getattr(result, "tool_calls", 0),
+        )
+    except Exception as e:  # noqa: BLE001
+        logger.error("agentic run failed: %s", e)
+        return AgenticResult(success=False, error=str(e))
+
+
+# ============================================================
+# 工具注册表 (default = 复用 mcp_tools)
+# ============================================================
+
+def _default_tool_registry() -> dict[str, Any]:
+    """复用 mcp_server 的工具定义。lazy import 避免循环依赖"""
+    try:
+        from ..database import mcp_tools  # type: ignore
+        registry: dict[str, Any] = {}
+        for getter in (
+            "get_connection_tools",
+            "get_table_analysis_tools",
+            "get_codegen_tools",
+            "get_springboot_project_tools",
+        ):
+            fn = getattr(mcp_tools, getter, None)
+            if fn is None:
+                continue
+            for tool in fn():
+                name = getattr(tool, "name", None) or (
+                    tool.get("name") if isinstance(tool, dict) else None
+                )
+                if name:
+                    registry[name] = tool
+        return registry
+    except Exception as e:  # noqa: BLE001
+        logger.warning("default tool registry build failed: %s", e)
+        return {}
+
+
+def _filter_tools(registry: dict[str, Any], allowed: Optional[list[str]]) -> dict[str, Any]:
+    if not allowed:
+        return registry
+    return {k: v for k, v in registry.items() if k in allowed}
diff --git a/tests/unit/test_agentic_runner.py b/tests/unit/test_agentic_runner.py
@@ -0,0 +1,105 @@
+"""Unit tests for server.agentic_runner."""
+import os
+
+import pytest
+
+from dbjavagenix.server.agentic_runner import (
+    AgenticConfig,
+    AgenticResult,
+    _filter_tools,
+    is_agentic_available,
+    run_agentic,
+)
+
+
+class TestAgenticConfig:
+    def test_minimal_valid(self):
+        c = AgenticConfig(goal="analyze schema")
+        assert c.goal == "analyze schema"
+        assert c.model == "claude-opus-4-7"
+        assert c.max_iterations == 20
+        assert c.allowed_tools is None
+        assert c.dry_run is False
+
+    def test_empty_goal_raises(self):
+        with pytest.raises(ValueError, match="goal"):
+            AgenticConfig(goal="")
+
+    def test_non_string_goal_raises(self):
+        with pytest.raises(ValueError):
+            AgenticConfig(goal=123)  # type: ignore
+
+    def test_max_iter_too_low(self):
+        with pytest.raises(ValueError, match="max_iterations"):
+            AgenticConfig(goal="x", max_iterations=0)
+
+    def test_max_iter_too_high(self):
+        with pytest.raises(ValueError):
+            AgenticConfig(goal="x", max_iterations=101)
+
+
+class TestAvailability:
+    def test_no_api_key_returns_false(self, monkeypatch):
+        monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+        ok, reason = is_agentic_available()
+        assert ok is False
+        assert "ANTHROPIC_API_KEY" in reason
+
+
+class TestFilterTools:
+    def test_none_passes_all(self):
+        r = {"a": 1, "b": 2}
+        assert _filter_tools(r, None) == r
+
+    def test_empty_list_passes_all(self):
+        r = {"a": 1, "b": 2}
+        assert _filter_tools(r, []) == r
+
+    def test_filter_to_subset(self):
+        r = {"a": 1, "b": 2, "c": 3}
+        assert _filter_tools(r, ["a", "c"]) == {"a": 1, "c": 3}
+
+    def test_unknown_name_silently_dropped(self):
+        r = {"a": 1}
+        assert _filter_tools(r, ["a", "nonexistent"]) == {"a": 1}
+
+
+class TestDryRun:
+    """dry_run 路径不需要真实 SDK,只验证配置和工具数量。"""
+
+    def test_dry_run_with_key_succeeds(self, monkeypatch):
+        # dry_run 走 is_agentic_available 前置,需要 key 才能进 dry_run 分支
+        # 但 SDK 可能没装。我们直接让 is_agentic_available mock 为 True 的
+        # 路径靠 monkeypatch 实现。
+        monkeypatch.setenv("ANTHROPIC_API_KEY", "fake-test-key")
+        # 如果 sdk 没装 is_agentic_available 仍返回 False,我们 monkeypatch 它
+        import dbjavagenix.server.agentic_runner as mod
+        monkeypatch.setattr(mod, "is_agentic_available", lambda: (True, ""))
+
+        config = AgenticConfig(goal="test goal", dry_run=True)
+        result = run_agentic(config, tool_registry_factory=lambda: {"tool_a": {}, "tool_b": {}})
+        assert result.success is True
+        assert result.iterations_used == 0
+        assert "test goal" in result.final_message
+        assert "tools=2" in result.final_message
+
+    def test_dry_run_respects_allowed_tools(self, monkeypatch):
+        import dbjavagenix.server.agentic_runner as mod
+        monkeypatch.setattr(mod, "is_agentic_available", lambda: (True, ""))
+
+        config = AgenticConfig(
+            goal="x", dry_run=True, allowed_tools=["tool_a"]
+        )
+        result = run_agentic(config, tool_registry_factory=lambda: {"tool_a": {}, "tool_b": {}})
+        assert result.success is True
+        assert "tools=1" in result.final_message
+
+
+class TestUnavailable:
+    def test_returns_error_result_when_unavailable(self, monkeypatch):
+        monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+        config = AgenticConfig(goal="x")
+        result = run_agentic(config)
+        assert result.success is False
+        assert result.error is not None
+        assert "unavailable" in result.error
