SDK Guide
Monstrum SDK Guide
Everything you need to automate and extend Monstrum programmatically using the Python SDK.
Table of Contents
- Overview
- Installation
- Authentication
- Managing Bots
- Triggering Tasks
- Working with Resources
- Building Custom Integrations
- Declarative Plugin Model
- Error Handling
- API Reference
Overview
Monstrum is an AI agent governance platform. You can interact with it through three interfaces:
- Web Dashboard — configure bots, resources, permissions, and credentials through the UI.
- Python SDK —
pip install monstrumfor programmatic access to all platform capabilities. - REST API — direct HTTP calls for any language or environment.
The SDK is a thin, typed wrapper around the REST API. Anything you can do in the dashboard, you can automate with the SDK.
Installation
pip install monstrum
Requires Python 3.10+. The SDK has minimal dependencies and supports both sync and async usage.
import monstrum
# Verify installation
print(monstrum.__version__)
Authentication
Every SDK call requires an API key. Generate one from the Monstrum dashboard under Settings > API Keys.
Setting Your API Key
Option 1: Environment variable (recommended)
export MONSTRUM_API_KEY="mst_your_api_key_here"
import monstrum
# Automatically reads MONSTRUM_API_KEY from environment
client = monstrum.Client()
Option 2: Explicit initialization
import monstrum
client = monstrum.Client(api_key="mst_your_api_key_here")
Option 3: Configuration file
# ~/.monstrum/config.toml
[default]
api_key = "mst_your_api_key_here"
api_base = "https://api.monstrumai.com"
API Key Scopes
When creating an API key in the dashboard, you can restrict its scope:
| Scope | Access |
|---|---|
bots:read | List and inspect bots |
bots:write | Create, update, and delete bots |
tasks:write | Trigger and manage tasks |
resources:read | List resources and their configuration |
resources:write | Create and configure resources |
audit:read | Query audit logs |
admin | Full access (use sparingly) |
Managing Bots
Bots are AI agents governed by the platform. Each bot has an identity, a set of permitted tools, and scoped access to resources.
List Bots
bots = client.bots.list()
for bot in bots:
print(f"{bot.id}: {bot.name} (status={bot.status})")
Create a Bot
bot = client.bots.create(
name="deploy-bot",
description="Handles staging deployments",
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"Created bot: {bot.id}")
Update a Bot
client.bots.update(
bot_id="bot_abc123",
allowed_operations=["deploy.read", "deploy.write", "deploy.rollback"],
)
Delete a Bot
client.bots.delete(bot_id="bot_abc123")
Triggering Tasks
A task is a single execution of a bot — one conversation turn or one autonomous run. You can trigger tasks programmatically and retrieve results.
Run a Task
task = client.tasks.run(
bot_id="bot_abc123",
input="Deploy service user-api to staging with image tag v2.4.1",
)
print(f"Task {task.id}: {task.status}")
print(f"Output: {task.output}")
Run a Task (Async)
import asyncio
import monstrum
async def main():
client = monstrum.AsyncClient()
task = await client.tasks.run(
bot_id="bot_abc123",
input="Check the status of all staging deployments",
)
print(task.output)
asyncio.run(main())
Stream a Task
For long-running tasks, stream results as they arrive:
for event in client.tasks.stream(
bot_id="bot_abc123",
input="Run the full integration test suite",
):
if event.type == "tool_call":
print(f" Tool: {event.tool_name}({event.params})")
elif event.type == "result":
print(f" Result: {event.data}")
elif event.type == "complete":
print(f"Done. Output: {event.output}")
Retrieve Past Tasks
# Get a specific task
task = client.tasks.get(task_id="task_xyz789")
# List recent tasks for a bot
tasks = client.tasks.list(bot_id="bot_abc123", limit=20)
for t in tasks:
print(f"{t.id} [{t.status}] {t.created_at}")
Working with Resources
Resources represent external services (GitHub, Jira, AWS, databases) that bots can access. Resources are configured with credentials and permissions in the dashboard, then bound to bots.
List Resources
resources = client.resources.list()
for r in resources:
print(f"{r.id}: {r.type} - {r.name}")
Get Resource Details
resource = client.resources.get(resource_id="res_github_prod")
print(f"Type: {resource.type}")
print(f"Tools: {[t.name for t in resource.tools]}")
Query Audit Logs
Every tool call made by every bot is logged. Query logs for compliance and debugging:
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}")
Building Custom Integrations
The SDK makes it straightforward to build automation on top of Monstrum. Here are common patterns.
Scheduled Bot Runs
import schedule
import time
def daily_report():
task = client.tasks.run(
bot_id="bot_reporter",
input="Generate the daily infrastructure health report",
)
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-Triggered Tasks
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"Investigate alert: {alert['title']}. Details: {alert['description']}",
metadata={"alert_id": alert["id"], "source": "pagerduty"},
)
return {"task_id": task.id}, 202
CI/CD Pipeline Integration
# In your deployment script
import sys
task = client.tasks.run(
bot_id="bot_deployer",
input=f"Deploy {sys.argv[1]} to production with image {sys.argv[2]}",
)
if task.status != "completed":
print(f"Deployment failed: {task.error}")
sys.exit(1)
print(f"Deployment successful: {task.output}")
Declarative Plugin Model
Beyond using bots through the SDK, you can build custom integrations that run on the platform. Monstrum uses a declarative plugin model: you define your integration as a YAML manifest, and the platform handles permissions, credential injection, UI rendering, and audit logging.
How It Works
A custom integration adds a new resource type to your Monstrum workspace. You declare:
| Declaration | Purpose |
|---|---|
tools[] | Tool definitions your bots can call |
scope_dimensions[] | Permission rules (auto-enforced) |
auth_methods[] | Supported credential flows |
credential_schema[] | Credential fields (auto-rendered in dashboard) |
config_schema[] | Configuration fields (auto-rendered in dashboard) |
Example: Jira Integration
Upload a manifest via the dashboard or CLI to register a custom integration:
name: jira
version: 1.0.0
description: Jira integration
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: "List issues from a Jira project."
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: "Project {value} is not authorized"
Once uploaded, the platform auto-generates the credential forms, enforces permissions through the scope dimensions, and makes the tools available to any bot with the appropriate resource binding.
Upload via CLI
pip install monstrum
monstrum integrations upload ./jira-integration.yaml
Upload via SDK
with open("jira-integration.yaml") as f:
manifest = f.read()
client.integrations.upload(manifest=manifest)
Error Handling
The SDK raises typed exceptions for all error conditions:
import monstrum
from monstrum.exceptions import (
AuthenticationError,
PermissionError,
NotFoundError,
RateLimitError,
MonstrumAPIError,
)
try:
task = client.tasks.run(bot_id="bot_abc123", input="Do something")
except AuthenticationError:
print("Invalid or expired API key")
except PermissionError as e:
print(f"Insufficient scope: {e.required_scope}")
except NotFoundError:
print("Bot not found")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after}s")
except MonstrumAPIError as e:
print(f"API error {e.status_code}: {e.message}")
API Reference
For the complete REST API reference and detailed SDK documentation, visit: