import requests
import json
import time
# Alert Responder Management Functions
def create_alert_responder(team_name, name, matching_criteria, api_key,
webhook_sources=None, slack_channel_id=None,
runbook=None, notification_integration_ids=None):
"""
Create a new alert responder.
Args:
team_name: Team name
name: Alert responder name
matching_criteria: Matching criteria dict (text_matches, slack_bot_app_user_id)
api_key: Organization API key
webhook_sources: Optional list of webhook sources (for webhook-based alerts)
slack_channel_id: Optional Slack channel ID (for Slack alerts)
runbook: Optional runbook configuration
notification_integration_ids: Optional notification integration IDs
Note: Must specify either webhook_sources OR slack_channel_id (not both)
"""
url = "https://api.tierzero.ai/api/v1/alert-responders"
headers = {
"X-TierZero-Org-Api-Key": api_key,
"Content-Type": "application/json"
}
payload = {
"team_name": team_name,
"name": name,
"matching_criteria": matching_criteria
}
if webhook_sources:
payload["webhook_sources"] = webhook_sources
if slack_channel_id:
payload["slack_channel_id"] = slack_channel_id
if runbook:
payload["runbook"] = runbook
if notification_integration_ids:
payload["notification_integration_ids"] = notification_integration_ids
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
def get_alert_responder(alert_id, api_key):
"""Get alert responder details"""
url = f"https://api.tierzero.ai/api/v1/alert-responders/{alert_id}"
headers = {
"X-TierZero-Org-Api-Key": api_key
}
response = requests.get(url, headers=headers)
response.raise_for_status()
return response.json()
def list_alert_responders(api_key, team_name=None):
"""List all alert responders, optionally filtered by team"""
url = "https://api.tierzero.ai/api/v1/alert-responders"
headers = {
"X-TierZero-Org-Api-Key": api_key
}
params = {}
if team_name:
params["team_name"] = team_name
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
return response.json()
def update_alert_responder(alert_id, api_key, name=None, matching_criteria=None,
webhook_sources=None, slack_channel_id=None,
runbook=None, notification_integration_ids=None):
"""
Update an existing alert responder.
Args:
alert_id: Alert responder Global ID
api_key: Organization API key
name: Optional new name
matching_criteria: Optional new matching criteria
webhook_sources: Optional new webhook sources (for webhook-based alerts only)
slack_channel_id: Optional new Slack channel ID (for Slack alerts only)
runbook: Optional new runbook
notification_integration_ids: Optional new notification integration IDs
Note: Cannot mix alert types - don't specify webhook_sources for Slack alerts
or slack_channel_id for webhook-based alerts
"""
url = f"https://api.tierzero.ai/api/v1/alert-responders/{alert_id}"
headers = {
"X-TierZero-Org-Api-Key": api_key,
"Content-Type": "application/json"
}
payload = {}
if name:
payload["name"] = name
if matching_criteria:
payload["matching_criteria"] = matching_criteria
if webhook_sources:
payload["webhook_sources"] = webhook_sources
if slack_channel_id:
payload["slack_channel_id"] = slack_channel_id
if runbook:
payload["runbook"] = runbook
if notification_integration_ids:
payload["notification_integration_ids"] = notification_integration_ids
response = requests.put(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
def enable_alert_responder(alert_id, api_key):
"""Enable an alert responder"""
url = f"https://api.tierzero.ai/api/v1/alert-responders/{alert_id}/enable"
headers = {
"X-TierZero-Org-Api-Key": api_key
}
response = requests.post(url, headers=headers)
response.raise_for_status()
return response.json()
def disable_alert_responder(alert_id, api_key):
"""Disable an alert responder"""
url = f"https://api.tierzero.ai/api/v1/alert-responders/{alert_id}/disable"
headers = {
"X-TierZero-Org-Api-Key": api_key
}
response = requests.post(url, headers=headers)
response.raise_for_status()
return response.json()
def delete_alert_responder(alert_id, api_key):
"""Delete an alert responder"""
url = f"https://api.tierzero.ai/api/v1/alert-responders/{alert_id}"
headers = {
"X-TierZero-Org-Api-Key": api_key
}
response = requests.delete(url, headers=headers)
response.raise_for_status()
return response.json()
def list_webhook_subscriptions(api_key):
"""List available webhook subscriptions"""
url = "https://api.tierzero.ai/api/v1/webhook-subscriptions"
headers = {
"X-TierZero-Org-Api-Key": api_key
}
response = requests.get(url, headers=headers)
response.raise_for_status()
return response.json()
def list_notification_integrations(api_key, kind=None):
"""
List available notification integrations.
Args:
api_key: Organization API key
kind: Optional filter by kind. Allowed values: DISCORD_WEBHOOK, SLACK_ALERT
"""
url = "https://api.tierzero.ai/api/v1/notification-integrations"
headers = {
"X-TierZero-Org-Api-Key": api_key
}
params = {}
if kind:
params["kind"] = kind
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
return response.json()
# Alert responder usage examples
# First, discover available webhook sources
webhook_subs = list_webhook_subscriptions("your-api-key")
print(f"Available webhook subscriptions: {len(webhook_subs['webhook_subscriptions'])}")
# Discover available notification integrations
notification_integrations = list_notification_integrations("your-api-key")
print(f"Available notification integrations: {len(notification_integrations['notification_integrations'])}")
# Create webhook-based alert responder (PagerDuty example)
webhook_alert = create_alert_responder(
team_name="Production",
name="Critical Error Alert",
matching_criteria={"text_matches": ["critical", "error", "fatal"]},
webhook_sources=[{"type": "PAGERDUTY", "remote_id": "PXXXXXX"}],
api_key="your-api-key",
runbook={
"investigation_prompt": "Analyze this alert and provide detailed root cause analysis with remediation steps",
"impact_and_severity_prompt": "Quick triage: assess severity and impact"
},
notification_integration_ids=["R3JhcGhRTE5vdGlmaWNhdGlvbkludGVncmF0aW9uOjEyMw=="]
)
print(f"Created webhook alert responder: {webhook_alert['url']}")
# Create Slack alert responder
slack_alert = create_alert_responder(
team_name="Production",
name="Database Alerts",
matching_criteria={
"text_matches": ["database", "connection", "timeout"],
"slack_bot_app_user_id": "B01234567" # Optional: filter by bot ID
},
slack_channel_id="C01234567",
api_key="your-api-key",
runbook={
"investigation_prompt": "Investigate database issues and provide remediation steps"
},
notification_integration_ids=["R3JhcGhRTE5vdGlmaWNhdGlvbkludGVncmF0aW9uOjEyMw=="]
)
print(f"Created Slack alert responder: {slack_alert['url']}")
# Get alert details
alert_details = get_alert_responder(webhook_alert["id"], "your-api-key")
print(f"Alert status: {alert_details['status']}")
# List all alert responders
all_alerts = list_alert_responders("your-api-key")
print(f"Total alert responders: {len(all_alerts['alert_responders'])}")
# Update webhook-based alert responder
updated = update_alert_responder(
webhook_alert["id"],
"your-api-key",
name="Updated Critical Alert",
matching_criteria={"text_matches": ["critical", "emergency"]}
)
print(f"Updated alert responder: {updated['name']}")
# Update Slack alert responder
updated_slack = update_alert_responder(
slack_alert["id"],
"your-api-key",
slack_channel_id="C98765432",
matching_criteria={
"text_matches": ["critical", "database"],
"slack_bot_app_user_id": "B99999999"
}
)
print(f"Updated Slack alert responder: {updated_slack['name']}")
# Disable alert responder temporarily
disable_alert_responder(webhook_alert["id"], "your-api-key")
print("Alert responder disabled")
# Re-enable it
enable_alert_responder(webhook_alert["id"], "your-api-key")
print("Alert responder enabled")
# Digest Management Functions
def create_digest(team_name, name, sections, schedule, api_key, slack_channel_id=None):
"""Create a new digest job"""
url = "https://api.tierzero.ai/api/v1/digests"
headers = {
"X-TierZero-Org-Api-Key": api_key,
"Content-Type": "application/json"
}
payload = {
"team_name": team_name,
"name": name,
"sections": sections,
"schedule": schedule
}
if slack_channel_id:
payload["slack_channel_id"] = slack_channel_id
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
def get_digest(digest_id, api_key):
"""Get digest details by digest ID"""
url = f"https://api.tierzero.ai/api/v1/digests/{digest_id}"
headers = {
"X-TierZero-Org-Api-Key": api_key
}
response = requests.get(url, headers=headers)
response.raise_for_status()
return response.json()
def update_digest(digest_id, api_key, name=None, sections=None, schedule=None, slack_channel_id=None):
"""Update an existing digest"""
url = f"https://api.tierzero.ai/api/v1/digests/{digest_id}"
headers = {
"X-TierZero-Org-Api-Key": api_key,
"Content-Type": "application/json"
}
payload = {}
if name:
payload["name"] = name
if sections:
payload["sections"] = sections
if schedule:
payload["schedule"] = schedule
if slack_channel_id:
payload["slack_channel_id"] = slack_channel_id
response = requests.put(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
def run_digest(digest_id, api_key, run_sync=False, scheduled_runtime=None):
"""
Manually run a digest job immediately or schedule it for later.
Args:
digest_id: Digest ID
api_key: Organization API key
run_sync: If True, runs synchronously and returns results
scheduled_runtime: Optional Unix timestamp (seconds) to schedule the run
Note: Notifications are always sent to configured channels/webhooks
"""
url = f"https://api.tierzero.ai/api/v1/digests/{digest_id}/run"
headers = {
"X-TierZero-Org-Api-Key": api_key,
"Content-Type": "application/json"
}
payload = {}
if run_sync:
payload["run_sync"] = run_sync
if scheduled_runtime:
payload["scheduled_runtime"] = scheduled_runtime
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
def delete_digest(digest_id, api_key):
"""Delete a digest job"""
url = f"https://api.tierzero.ai/api/v1/digests/{digest_id}"
headers = {
"X-TierZero-Org-Api-Key": api_key
}
response = requests.delete(url, headers=headers)
response.raise_for_status()
return response.json()
# Scheduled Actions Functions
def create_scheduled_action(
question,
scheduled_runtimes,
api_key,
slack_channel_ids=None,
discord_webhook_urls=None
):
"""
Create a scheduled action that runs a TierZero agent at specific times.
Args:
question: Question/prompt to send to TierZero agent
scheduled_runtimes: List of Unix timestamps (seconds) when action should run
api_key: Organization API key
slack_channel_ids: Optional list of Slack channel IDs for notifications
discord_webhook_urls: Optional list of Discord webhook URLs for notifications
Note:
Notification targets are optional. If not provided, query results using GET /api/v1/interactions/{interaction_id}.
Returns:
Dictionary with:
- interactions (list of dicts): One entry per scheduled runtime
- status (str): SCHEDULED or ACCEPTED
Each interaction dict contains: interaction_id, scheduled_runtime, and url.
"""
url = "https://api.tierzero.ai/api/v1/interactions"
headers = {
"X-TierZero-Org-Api-Key": api_key,
"Content-Type": "application/json"
}
payload = {
"question": question,
"scheduled_runtimes": scheduled_runtimes
}
# Build nested notifications structure
notifications = {}
if slack_channel_ids:
notifications["slack"] = {"slack_channel_ids": slack_channel_ids}
if discord_webhook_urls:
notifications["discord"] = {"discord_webhook_urls": discord_webhook_urls}
if notifications:
payload["notifications"] = notifications
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
# Scheduled Action usage example
scheduled_action_result = create_scheduled_action(
question="Analyze system performance metrics from the last 7 days and identify any anomalies or trends",
scheduled_runtimes=[1735689600, 1735776000], # Unix timestamps for when to run
api_key="your-api-key",
slack_channel_ids=["C01234567"]
)
print(f"Status: {scheduled_action_result['status']}") # "SCHEDULED"
print(f"Created {len(scheduled_action_result['interactions'])} scheduled interactions:")
for interaction in scheduled_action_result['interactions']:
print(f" - ID: {interaction['interaction_id']}")
print(f" Scheduled at: {interaction['scheduled_runtime']}")
print(f" URL: {interaction['url']}")
# Store all interaction_ids for tracking individual executions
interaction_ids = [i['interaction_id'] for i in scheduled_action_result['interactions']]
# Digest usage example
digest_result = create_digest(
team_name="Engineering",
name="Daily Production Health",
sections=[
{
"section": "Error Summary",
"prompt": "Summarize production errors from the last 24 hours"
},
{
"section": "Performance Metrics",
"prompt": "Report on API response times and throughput"
}
],
schedule="0 9 * * 1-5",
api_key="your-api-key",
slack_channel_id="C01234567"
)
print(f"Created digest: {digest_result['url']}")
# Run the digest immediately and asynchronously
run_digest(digest_result["id"], "your-api-key")
# Run the digest synchronously to get immediate results
run_result = run_digest(
digest_result["id"],
"your-api-key",
run_sync=True
)
for section_result in run_result.get("results", []):
print(f"\n{section_result['section']}:")
print(section_result['result'])
# Schedule the digest to run at a specific time (e.g., tomorrow at 9 AM)
import time
from datetime import datetime, timedelta
tomorrow_9am = datetime.now() + timedelta(days=1)
tomorrow_9am = tomorrow_9am.replace(hour=9, minute=0, second=0, microsecond=0)
scheduled_result = run_digest(
digest_result["id"],
"your-api-key",
scheduled_runtime=int(tomorrow_9am.timestamp())
)
print(f"Digest scheduled for: {datetime.fromtimestamp(scheduled_result['scheduled_runtime'])}")
# Interaction Management Functions
def create_interaction_sync(question, api_key, context=None, user_id=None, use_tools=True):
"""Create a synchronous interaction"""
url = "https://api.tierzero.ai/api/v1/interactions"
headers = {
"X-TierZero-Org-Api-Key": api_key,
"Content-Type": "application/json"
}
payload = {"question": question, "use_tools": use_tools}
if context:
payload["context"] = context
if user_id:
payload["user_id"] = user_id
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
def create_interaction_async(question, api_key, context=None, user_id=None, use_tools=True):
"""Create an asynchronous interaction"""
url = "https://api.tierzero.ai/api/v1/interactions"
headers = {
"X-TierZero-Org-Api-Key": api_key,
"Content-Type": "application/json"
}
payload = {"question": question, "async": True, "use_tools": use_tools}
if context:
payload["context"] = context
if user_id:
payload["user_id"] = user_id
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()
def get_interaction_results(interaction_id, api_key):
"""Get interaction results by ID"""
url = f"https://api.tierzero.ai/api/v1/interactions/{interaction_id}"
headers = {
"X-TierZero-Org-Api-Key": api_key,
"Accept": "application/json"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
return response.json()
def wait_for_completion(interaction_id, api_key, timeout=60, poll_interval=2):
"""Poll for interaction completion"""
start_time = time.time()
while time.time() - start_time < timeout:
result = get_interaction_results(interaction_id, api_key)
if result.get("status") == "COMPLETED":
return result
elif result.get("status") == "FAILED":
raise Exception(f"Interaction failed: {interaction_id}")
print(f"Status: {result.get('status')}, waiting...")
time.sleep(poll_interval)
raise TimeoutError(f"Interaction {interaction_id} did not complete within {timeout} seconds")
# Synchronous usage
result = create_interaction_sync(
"What are the top error patterns in production today?",
"your-api-key",
user_id="analyst123"
)
print(result["content"])
# Asynchronous usage
async_result = create_interaction_async(
"Analyze the performance impact of our latest deployment",
"your-api-key",
user_id="analyst123"
)
print(f"Started async interaction: {async_result['interaction_id']}")
# Wait for completion
final_result = wait_for_completion(async_result["interaction_id"], "your-api-key")
print(final_result["content"])
# Request without tools (knowledge-only response)
knowledge_result = create_interaction_sync(
"What is our standard deployment process?",
"your-api-key",
use_tools=False
)
print(knowledge_result["content"])
