opspilot-assistant 0.2.0

AI-driven operations assistant system

OpsPilot

AI 驱动的运维告警处置系统。接收告警 → LLM 分析根因 → 执行诊断工具 → 危险操作人工审批 → 生成结构化报告，全程审计。

架构分层

OpsPilot 分为三层，依赖严格单向（tui → cli → app → core）：

层	包	职责	安装
Core（可嵌入 SDK）	opspilot	纯管线：去重/队列/会话/Agent 循环/工具/持久化抽象。不依赖 FastAPI / PyYAML / 文件 I/O	默认
App（standalone 程序）	opspilot.app	OpsPilot 应用类 + 配置解析 + ConfigStore 版本管理 + Webhook/Puller 接入 + Control API（REST + SSE）+ FastAPI 服务	[app]
CLI	opspilot.cli	Typer 命令行：远程管理 + opspilot serve 本地启动 + Rich 表格/JSON 输出	[cli]
TUI	opspilot.cli.tui	Textual 仪表盘：实时告警/会话/审批面板	[tui]

一个 OpsPilot 实例 = 一个完整应用（一个服务器）。嵌入方只需管线时可单独使用 core；需要 standalone 服务时使用 app。

告警源 (Prometheus / Grafana / 自定义)
        │  Webhook 推送 / Pull 拉取 / 归一化
        ▼
   ┌──────────┐
   │ 调度层   │  去重 / 优先级队列 / Worker 池
   └────┬─────┘
        │ AlertEvent
        ▼
   ┌──────────┐
   │ 决策层   │  Agent 循环: 上下文 → LLM → 工具 → 安全检查 → 审批
   └────┬─────┘
        ▼
   ┌──────────┐
   │ 处置层   │  K8s / Shell / 数据库 / 通知 / 知识检索 / MCP
   └────┬─────┘
        ▼
   ┌──────────┐
   │ 存储层   │  告警 / 会话 / 分析 / 工具调用 / 审批 / 审计日志
   └──────────┘

安装

# 全量安装（CLI + TUI + 服务端）
pip install opspilot-assistant[all]

# 按需安装
pip install opspilot-assistant          # 仅 core SDK（嵌入方）
pip install opspilot-assistant[app]     # core + 服务端（程序化使用）
pip install opspilot-assistant[cli]     # core + CLI（含 serve）
pip install opspilot-assistant[tui]     # core + CLI + TUI 仪表盘

开发环境：

git clone <repo-url> && cd opspilot
uv sync --extra all

最小配置（config.yaml）：

server:
  auth_token: "${OPSPILOT_TOKEN}"   # Control API 鉴权 token（为空则不校验，仅限开发）
providers:
  mimo:
    model: "mimo-v2.5-pro"
    api_keys: ["${MIMO_KEY}"]
    base_url: "https://token-plan-cn.xiaomimimo.com/v1"
actions:
  tools:
    builtins: {}
projects:
  - name: my-project
    llm:
      primary: mimo
    ingestion:
      webhook:
        enabled: true
        secret: "${WEBHOOK_SECRET}"

方式一：启动 standalone HTTP 服务

create_app(ops) 装配一个 FastAPI 应用，挂载 Webhook + Control API + SSE，全部共用同一端口：

import asyncio
import uvicorn
from opspilot.app import OpsPilot, create_app

async def main():
    async with OpsPilot(open("config.yaml").read()) as op:
        app = create_app(op)
        await uvicorn.Server(uvicorn.Config(app, port=8000)).serve()

asyncio.run(main())

或使用 CLI：

opspilot serve --config config.yaml

服务启动后：

• POST /api/v1/webhook/{project} — 接收告警推送（X-Webhook-Secret 鉴权）
• GET /api/v1/sessions / POST /api/v1/sessions/{id}/approve — 控制 API（Authorization: Bearer 鉴权）
• GET /api/v1/events/stream — SSE 实时事件流
• GET /api/v1/health / /api/v1/ready — 存活/就绪探针
• /docs — Swagger UI（OpenAPI 自动生成）

方式二：程序化嵌入（不开端口）

import asyncio
from opspilot.app import OpsPilot

async def main():
    async with OpsPilot(open("config.yaml").read()) as op:
        result = await op.ingest([event])       # 推入已归一化的告警
        await op.approve(session_id, "ok")      # 审批危险操作
        await op.cancel_session(session_id)     # 取消会话

asyncio.run(main())

只需管线（不要 FastAPI/uvicorn）时可直接用 core SDK：

from opspilot import AlertEvent, Severity, AlertStatus

技术栈

组件	选型	说明
语言	Python 3.14
包管理	uv	uv sync / uv run
Web 框架	FastAPI + uvicorn	Control API + SSE + Webhook
Agent 引擎	手写 async 循环	上下文压缩 + 非阻塞审批 + 持久化（无 LangGraph）
LLM 抽象	LangChain	多模型 + 密钥轮换 + provider 降级
CLI	Typer + Rich	运维友好的表格/JSON 输出
TUI	Textual	实时终端仪表盘（[tui] extra）
数据库	SQLAlchemy + SQLite（可换）	引擎按连接 URL 前缀注册，单文件部署
调度	APScheduler	Pull 模式定时拉取
MCP	langchain-mcp-adapters	远程工具服务器扩展

文档

• 配置参考 — 完整配置项说明
• Control API — REST + SSE 端点规范
• 集成指南 — 告警源 / 通知 / K8s / 数据库 / MCP / Shell
• 需求规格 — 系统设计与功能定义

包结构

src/
  core/        opspilot           — 可嵌入核心 SDK（管线 + 抽象 + 数据模型）
  app/         opspilot.app       — standalone 程序层（OpsPilot + HTTP + 配置 + 接入）
  builtin/     opspilot.builtin   — 内置实现（工具/归一化器/拉取器/引擎/通知）
  cli/         opspilot.cli       — CLI 命令行（Typer + Rich）
  cli/tui/     opspilot.cli.tui   — TUI 仪表盘（Textual）