SDK 指南

Monstrum SDK 指南

通过 Python SDK 以编程方式自动化和扩展 Monstrum 所需的一切。

目录

概览

Monstrum 是一个 AI Agent 治理平台，提供三种交互方式：

  1. Web 控制台 — 通过界面配置 Bot、资源、权限和凭据。
  2. Python SDK — 通过 pip install monstrum 以编程方式访问平台的所有功能。
  3. REST API — 适用于任何语言或环境的直接 HTTP 调用。

SDK 是对 REST API 的轻量级类型化封装。所有在控制台中能完成的操作，都可以通过 SDK 实现自动化。

安装

pip install monstrum

需要 Python 3.10+。SDK 依赖极少，同时支持同步和异步用法。

import monstrum

# 验证安装
print(monstrum.__version__)

认证

所有 SDK 调用都需要 API Key。在 Monstrum 控制台的 设置 > API Keys 中生成。

设置 API Key

方式一：环境变量（推荐）

export MONSTRUM_API_KEY="mst_your_api_key_here"
import monstrum

# 自动从环境变量读取 MONSTRUM_API_KEY
client = monstrum.Client()

方式二：显式初始化

import monstrum

client = monstrum.Client(api_key="mst_your_api_key_here")

方式三：配置文件

# ~/.monstrum/config.toml
[default]
api_key = "mst_your_api_key_here"
api_base = "https://api.monstrumai.com"

API Key 权限范围

在控制台创建 API Key 时可以限制其权限范围：

范围访问权限
bots:read列出和查看 Bot
bots:write创建、更新和删除 Bot
tasks:write触发和管理任务
resources:read列出资源及其配置
resources:write创建和配置资源
audit:read查询审计日志
admin完整访问权限（请谨慎使用）

管理 Bot

Bot 是受平台治理的 AI Agent，每个 Bot 拥有独立的身份、允许使用的工具集合，以及对资源的受限访问权限。

列出 Bot

bots = client.bots.list()
for bot in bots:
    print(f"{bot.id}: {bot.name} (status={bot.status})")

创建 Bot

bot = client.bots.create(
    name="deploy-bot",
    description="负责预发布环境的部署",
    model="gpt-4o",
    allowed_operations=["deploy.read", "deploy.write"],
    resource_bindings=[
        {
            "resource_id": "res_aws_staging",
            "allowed_tools": ["aws_ecs_deploy", "aws_ecs_status"],
            "scope_constraints": {
                "clusters": ["staging-*"],
                "regions": ["us-west-2"],
            },
        }
    ],
)
print(f"已创建 Bot: {bot.id}")

更新 Bot

client.bots.update(
    bot_id="bot_abc123",
    allowed_operations=["deploy.read", "deploy.write", "deploy.rollback"],
)

删除 Bot

client.bots.delete(bot_id="bot_abc123")

触发任务

任务是 Bot 的一次执行，对应一轮对话或一次自主运行。你可以通过 SDK 触发任务并获取结果。

运行任务

task = client.tasks.run(
    bot_id="bot_abc123",
    input="将 user-api 服务部署到预发布环境，镜像标签为 v2.4.1",
)

print(f"任务 {task.id}: {task.status}")
print(f"输出: {task.output}")

异步运行任务

import asyncio
import monstrum

async def main():
    client = monstrum.AsyncClient()
    task = await client.tasks.run(
        bot_id="bot_abc123",
        input="检查所有预发布环境的部署状态",
    )
    print(task.output)

asyncio.run(main())

流式获取任务结果

对于长时间运行的任务，可以流式获取结果：

for event in client.tasks.stream(
    bot_id="bot_abc123",
    input="运行完整的集成测试套件",
):
    if event.type == "tool_call":
        print(f"  工具: {event.tool_name}({event.params})")
    elif event.type == "result":
        print(f"  结果: {event.data}")
    elif event.type == "complete":
        print(f"完成。输出: {event.output}")

查询历史任务

# 获取特定任务
task = client.tasks.get(task_id="task_xyz789")

# 列出某个 Bot 的近期任务
tasks = client.tasks.list(bot_id="bot_abc123", limit=20)
for t in tasks:
    print(f"{t.id} [{t.status}] {t.created_at}")

使用资源

资源代表 Bot 可以访问的外部服务（GitHub、Jira、AWS、数据库等）。资源在控制台中配置凭据和权限后绑定到 Bot。

列出资源

resources = client.resources.list()
for r in resources:
    print(f"{r.id}: {r.type} - {r.name}")

获取资源详情

resource = client.resources.get(resource_id="res_github_prod")
print(f"类型: {resource.type}")
print(f"工具: {[t.name for t in resource.tools]}")

查询审计日志

Bot 的每一次工具调用都会被记录，可用于合规审查和问题排查：

logs = client.audit.list(
    bot_id="bot_abc123",
    resource_id="res_github_prod",
    start_time="2026-02-01T00:00:00Z",
    end_time="2026-02-15T00:00:00Z",
)
for entry in logs:
    print(f"{entry.timestamp} | {entry.tool_name} | {entry.status}")

构建自定义集成

SDK 使得在 Monstrum 之上构建自动化流程变得简单直接，以下是一些常见模式。

定时运行 Bot

import schedule
import time

def daily_report():
    task = client.tasks.run(
        bot_id="bot_reporter",
        input="生成每日基础设施健康报告",
    )
    if task.status == "completed":
        send_to_slack(task.output)

schedule.every().day.at("09:00").do(daily_report)
while True:
    schedule.run_pending()
    time.sleep(60)

Webhook 触发任务

from flask import Flask, request

app = Flask(__name__)

@app.route("/webhook/alert", methods=["POST"])
def handle_alert():
    alert = request.json
    task = client.tasks.run(
        bot_id="bot_incident",
        input=f"排查告警: {alert['title']}。详情: {alert['description']}",
        metadata={"alert_id": alert["id"], "source": "pagerduty"},
    )
    return {"task_id": task.id}, 202

CI/CD 流水线集成

# 在部署脚本中使用
import sys

task = client.tasks.run(
    bot_id="bot_deployer",
    input=f"将 {sys.argv[1]} 部署到生产环境，镜像为 {sys.argv[2]}",
)

if task.status != "completed":
    print(f"部署失败: {task.error}")
    sys.exit(1)

print(f"部署成功: {task.output}")

声明式插件模型

除了通过 SDK 使用 Bot 之外，你还可以构建运行在平台上的自定义集成。Monstrum 采用声明式插件模型：你将集成定义为一个 YAML 清单，平台自动处理权限执行、凭据注入、UI 渲染和审计日志。

工作原理

自定义集成会向你的 Monstrum 工作区添加一个新的资源类型。你需要声明以下内容：

声明用途
tools[]Bot 可调用的工具定义
scope_dimensions[]权限规则（自动执行）
auth_methods[]支持的凭据流程
credential_schema[]凭据字段（在控制台自动渲染）
config_schema[]配置字段（在控制台自动渲染）

示例：Jira 集成

通过控制台或 CLI 上传清单文件来注册自定义集成：

name: jira
version: 1.0.0
description: Jira 集成

resource_type:
  id: jira
  name: Jira

  credential_schema:
    - field: api_token
      type: secret
      required: true
    - field: email
      type: string
      required: true

  config_schema:
    - field: api_base
      type: url
      required: true

  tools:
    - name: jira_list_issues
      description: "列出 Jira 项目的 Issue。"
      operation: issue.read
      input_schema:
        type: object
        properties:
          project:
            type: string
          status:
            type: string
          max_results:
            type: integer
            default: 50
        required: [project]

  scope_dimensions:
    - key: projects
      param_paths: [project]
      match_mode: pattern
      error_template: "项目 {value} 不在授权范围内"

上传后，平台会自动生成凭据表单，通过 scope dimensions 执行权限控制，并将工具提供给拥有相应资源绑定的所有 Bot。

通过 CLI 上传

pip install monstrum
monstrum integrations upload ./jira-integration.yaml

通过 SDK 上传

with open("jira-integration.yaml") as f:
    manifest = f.read()

client.integrations.upload(manifest=manifest)

错误处理

SDK 为所有错误条件提供了类型化异常：

import monstrum
from monstrum.exceptions import (
    AuthenticationError,
    PermissionError,
    NotFoundError,
    RateLimitError,
    MonstrumAPIError,
)

try:
    task = client.tasks.run(bot_id="bot_abc123", input="执行操作")
except AuthenticationError:
    print("API Key 无效或已过期")
except PermissionError as e:
    print(f"权限不足: {e.required_scope}")
except NotFoundError:
    print("Bot 不存在")
except RateLimitError as e:
    print(f"请求频率超限，{e.retry_after} 秒后重试")
except MonstrumAPIError as e:
    print(f"API 错误 {e.status_code}: {e.message}")

API 参考

完整的 REST API 参考和 SDK 详细文档请访问：

在 GitHub 上编辑此页