把 AI Agent 做成可扩展系统：OpsPilot 的配置化、插件化与安全重构

很多 AI Agent 项目第一版都能很快跑起来：写几个 prompt，接一个模型，注册几组工具，让模型根据上下文决定要不要调用工具。这个阶段最重要的是验证想法，代码里有一些 mock、硬编码和临时分支都可以接受。

但 OpsPilot 的目标是运维辅助系统，不是一次性 demo。它要面对 Prometheus、Grafana、Kubernetes、数据库、Shell、通知渠道、MCP 服务和用户自定义工具。如果工具来源不统一，配置模型不稳定，安全边界不清楚，项目很快会从“能跑”变成“很难改”。

这篇文章复盘的是 OpsPilot 最近一轮工程化重构：把 AI Agent 从一个能调用工具的流程，整理成可配置、可扩展、可控风险的系统。

第一版的问题

第一版最直接的问题是工具和核心流程之间的边界不够清楚。

例如，决策层里曾经出现过 mock 工具，处置层里容易出现针对某类工具的硬编码分支，配置结构也更像临时参数集合。短期看，这些写法可以快速验证“LLM 能不能分析告警并调用工具”；长期看，它们会带来几个后果：

• core 层知道太多具体实现，后续新增工具会污染核心流程。
• 工具来源不统一，内置工具、用户工具、MCP 工具很难放进同一条加载链路。
• 配置覆盖关系不清晰，多项目场景下很难表达“全局启用，但某个项目禁用某些工具”。
• 安全策略容易散落在各个工具里，无法统一回答某次工具调用为什么被允许或拦截。

因此这轮重构的目标很明确：把机制和策略分离，把工具加载配置化，把风险控制前置到 Agent 工作流里。

重构一：core 和 builtin 分离

OpsPilot 现在的包结构是：

src/core 只放框架机制：配置模型、抽象基类、注册表、调度器、LangGraph 工作流、存储接口。src/builtin 放内置策略：具体 normalizer、puller 和 tool provider。

这件事看起来只是移动文件，但它影响的是后续所有扩展方式。核心层不再需要知道“Prometheus normalizer 怎么解析 payload”，也不需要知道“Shell 工具有哪些参数”。它只需要知道 normalizer 能把 payload 转成 AlertEvent，tool provider 能创建 LangChain BaseTool。

这种分离让扩展点变得更自然：新增一个工具时，不应该去改 Agent 图；新增一个告警源时，不应该去改调度器；新增一个数据源 Puller 时，不应该影响 webhook 接入。

重构二：ToolProvider 插件机制

处置层的关键抽象是 ToolProvider。每个 provider 有一个唯一的 tool_id，并实现异步的 create_tools()：

class ToolProvider(ABC):
    tool_id: str = ""

    @abstractmethod
    async def create_tools(self, config: dict[str, Any]) -> list[BaseTool]:
        """Create and return tools from the given config dict."""

    def validate_config(self, config: dict[str, Any]) -> None:
        """Optional: validate config before create_tools is called."""

内置工具通过装饰器注册到全局注册表：

_PROVIDER_REGISTRY: dict[str, type[ToolProvider]] = {}

def register_provider(cls: type[ToolProvider]) -> type[ToolProvider]:
    if not getattr(cls, "tool_id", ""):
        raise ValueError(f"{cls.__name__} must define a non-empty tool_id")
    _PROVIDER_REGISTRY[cls.tool_id] = cls
    return cls

这让 Shell、K8s、Notify、Database 都可以作为普通 provider 存在。核心加载流程不需要写 if shell then ... elif k8s then ... 这类分支，而是从注册表里找 provider，再让 provider 自己创建工具。

工具装配统一收敛到 build_tools()：

async def build_tools(
    actions_config: ActionsConfig | None = None,
    project_actions: ProjectActionsConfig | None = None,
) -> ToolBundle:
    if actions_config is None:
        return ToolBundle()

    proj = project_actions or ProjectActionsConfig()
    tools: list[BaseTool] = []
    safety_overrides: dict[str, str] = {}

    for provider_name, prov_cfg in actions_config.tools.items():
        if not prov_cfg.enabled:
            continue
        if provider_name == "builtins":
            await _load_builtins(prov_cfg, proj, tools)
        else:
            await _load_user_provider(provider_name, prov_cfg, proj, tools, safety_overrides)

    for server_name, mcp_cfg in actions_config.mcp.items():
        await _load_mcp_server(server_name, mcp_cfg, proj, tools, safety_overrides)

    return ToolBundle(tools=tools, safety_overrides=safety_overrides)

这个函数的职责很单一：读全局 actions 配置，合并项目级覆盖，加载静态 provider 和动态 MCP server，最后返回一个 ToolBundle。Agent 图拿到的只是工具列表和安全级别覆盖，不关心工具来自哪里。

重构三：配置从列表式走向字典式

这轮重构里一个很重要的变化，是把工具配置改成字典结构。顶层 actions.tools 和 actions.mcp 以 provider/server 名称作为 key：

actions:
  tools:
    builtins:
      disabled_tools: ["shell"]
    my-tools:
      path: "/home/ops/my-tools"
      enabled: true
      config:
        email:
          user: "me"
  mcp:
    my-mcp:
      transport: "http"
      url: "http://localhost:9500/mcp"
      enabled_tools: ["read_file"]

projects:
  - name: my-project
    actions:
      disabled_tools: ["my-mcp:*"]
      config:
        "my-tools:email":
          user: "prod-ops"

这个格式解决了两个问题。

第一，provider 和 MCP server 的身份变清楚了。my-tools、my-mcp 不只是列表里的一个元素，而是后续命名空间、过滤和覆盖的基础。

第二，全局配置和项目级配置的职责分开了。全局 actions 定义“系统有哪些工具来源”，项目级 actions 只定义差异：禁用哪些工具、覆盖哪些工具配置。这样多项目场景下不需要复制一大份工具配置，也不会让项目配置变成另一个全局配置。

配置模型本身也尽量保持简单：

class ActionsConfig(BaseModel):
    tool_paths: list[str] = Field(default_factory=list)
    tools: dict[str, ToolProviderConfig] = Field(default_factory=dict)
    mcp: dict[str, MCPServerConfig] = Field(default_factory=dict)

class ProjectActionsConfig(BaseModel):
    disabled_tools: list[str] = Field(default_factory=list)
    config: dict[str, dict[str, Any]] = Field(default_factory=dict)

项目级覆盖不在配置加载时提前合并，而是在 build_tools() 运行时合并。这样做的好处是，每个项目可以根据自己的上下文拿到不同的工具集合，而不需要复制或修改全局配置对象。

重构四：MCP 命名空间和工具过滤

MCP 带来的一个典型问题是工具名冲突。两个 MCP server 都可能暴露 read_file，内置工具和用户工具也可能重名。如果直接把工具扁平化交给 LLM，冲突很难定位。

OpsPilot 的处理方式是给 MCP 工具自动加命名空间：

for tool in raw_tools:
    original_name = tool.name
    namespaced = f"{server_name}:{original_name}"

    if not _is_tool_enabled(original_name, mcp_cfg.enabled_tools, mcp_cfg.disabled_tools):
        continue

    if not _is_tool_enabled(namespaced, [], proj.disabled_tools):
        continue

    tool.name = namespaced
    tools.append(tool)

这样 my-mcp 里的 read_file 会变成 my-mcp:read_file。项目配置里也可以直接写 my-mcp:* 禁用整个 server 的工具，或者用 enabled_tools 只开放其中几个工具。

过滤逻辑本身是统一的：

def _is_tool_enabled(
    tool_name: str,
    enabled_tools: list[str],
    disabled_tools: list[str],
) -> bool:
    if enabled_tools:
        if not any(fnmatch.fnmatch(tool_name, p) for p in enabled_tools):
            return False

    if disabled_tools:
        if any(fnmatch.fnmatch(tool_name, p) for p in disabled_tools):
            return False

    return True

白名单先判断，黑名单后判断，并支持 * 通配符。这个小函数让 provider 级过滤、MCP server 级过滤、项目级过滤可以复用同一套规则，避免每个加载分支各写一套判断。

安全设计：自动化边界必须清楚

AI 运维系统里最危险的不是模型不会调用工具，而是模型太容易调用工具。

OpsPilot 把工具调用分为三类：safe、dangerous、blocked。安全检查由 SafetyChecker 统一完成，而不是散落在每个节点里：

class SafetyLevel(str, Enum):
    SAFE = "safe"
    DANGEROUS = "dangerous"
    BLOCKED = "blocked"

class SafetyChecker:
    def check_tool_call(self, tool_name: str, arguments: dict[str, Any]) -> SafetyVerdict:
        level = self._tool_levels.get(tool_name, self._default_level)

        if level == SafetyLevel.DANGEROUS:
            command = arguments.get("command", "")
            if command:
                for pattern in self._blocked_shell:
                    if pattern.search(command):
                        return SafetyVerdict(
                            tool_name=tool_name,
                            level=SafetyLevel.BLOCKED,
                            reason=f"Command matches blocked pattern: {pattern.pattern}",
                        )

        return SafetyVerdict(tool_name=tool_name, level=level)

Shell 命令会检查阻断正则，SQL 类参数也有单独的黑名单匹配。dangerous 工具不会直接执行，而是进入 LangGraph 的审批节点。审批通过才执行，审批拒绝则回到 LLM 分析节点，让模型尝试其他方案。

这里的关键不是写了几个正则，而是安全判断的位置。工具风险在进入 execute_tools 之前完成，Agent 图的控制流保证了危险操作必须经过审批。也就是说，“能不能执行”是系统规则，不是模型自己说了算。

测试和工程化

这类重构很容易看起来只是“改目录、改配置、改名字”，但真正的风险是行为漂移。比如工具过滤顺序变了，某个项目突然多拿到危险工具；MCP 命名空间改了，安全配置匹配不到；Session 恢复时配置没有被正确反序列化。

所以 OpsPilot 目前按模块建立了测试：ingestion、dispatcher、agent、action、storage、cli 都有对应用例。调度层测试去重、队列和 worker；action 层测试 provider、registry、Shell、K8s、Notify、DB；agent 层测试 graph、context、LLM 适配和 safety；storage 层测试 ORM 和 repository。

这些测试的价值不是追求覆盖率数字，而是让重构可以继续发生。只要接口和行为被测试固定下来，后续新增工具、调整配置模型、接入更多 MCP server，就不会完全依赖手工验证。

复盘

这轮重构之后，OpsPilot 的处置链路变成了几个清楚的问题：

• 工具有哪些来源？由 actions.tools 和 actions.mcp 描述。
• 工具怎么被加载？由 ToolProvider、注册表和 MCP adapter 处理。
• 某个项目能用哪些工具？由项目级 disabled_tools 和 config 覆盖决定。
• 工具名如何避免冲突？MCP 工具使用 <server_name>:<tool_name> 命名空间。
• 工具调用是否安全？由 SafetyChecker 和 LangGraph 审批节点统一控制。

这也是我对 AI Agent 工程化的一点体会：真正难的不是让模型调用工具，而是让工具来源、配置覆盖、安全边界和运行状态都可解释、可测试、可扩展。

对于运维系统来说，自动化不是目的，可控的自动化才是目的。LLM 可以提高分析效率，但系统必须清楚地知道它能看什么、能调用什么、什么时候需要停下来等人审批。