跳转到主要内容

Documentation Index

Fetch the complete documentation index at: https://docs.olostep.com/llms.txt

Use this file to discover all available pages before exploring further.

通过 Olostep /v1/monitors 端点,你可以创建在固定时间表上运行的持久监控器,检测页面更改,并通过电子邮件、webhook 或短信通知你。
  • 从自然语言 query 创建监控器
  • 使用自然语言时间表运行检查
  • 通过 emailwebhook_urlphone_number (SMS) 发送警报
  • 列出、检查、更新、暂停、恢复和删除监控器
  • 从私人快照中读取监控器快照事件
默认情况下,每次监控器运行都会捕获被监控页面的完整快照——即该时刻页面当前状态的完整图像。如果你希望监控器仅显示运行之间的新内容或更改(增量),而不是完整状态,你必须在 query 中直接表达这一意图。

安装

# pip install requests

import requests

创建监控器

使用 POST /v1/monitors 创建监控器。端点会验证你的输入,配置监控器,创建其内部时间表,并返回一个活动的监控器对象。
  • query 是必需的。如果你的 query 包含一个 URL,监控器将专注于该输入 URL。默认情况下,每次运行都会获取被监控对象的完整快照——即页面当前状态的图像。如果你希望监控器仅跟踪运行之间的新内容或更改(例如,仅新博客文章或仅价格下降),你必须在 query 中明确指定;否则每次运行都会捕获完整快照。
  • frequency 接受自然语言时间表指令,如果没有提供时区,默认使用 UTC。
  • 必须有一个通知目标:emailwebhook_urlphone_number (SMS)。
  • 当你需要结构化提取结果时使用 output_schema。(可选)

示例请求

import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"

payload = {
    "query": "Track changes in https://example.com/products/widget-pro pricing and stock information",
    "frequency": "every day at 9am America/Los_Angeles",
    "email": "alerts@example.com",
    "output_schema": {
        "type": "object",
        "properties": {
            "title": {"type": "string"},
            "published_at": {"type": "string"}
        },
        "required": ["title"]
    },
    "metadata": {
        "product_id": "widget-pro",
        "team": "growth"
    }
}

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

response = requests.post(f"{API_URL}/monitors", headers=headers, json=payload)
print(response.status_code)
print(json.dumps(response.json(), indent=2))

响应

成功创建监控器返回 HTTP 202 和一个监控器对象:
{
  "id": "monitor_n8q2x4m1ak",
  "object": "monitor",
  "created_at": 1777983215,
  "updated_at": 1777983222,
  "fda_id": "fda_n8q2x4m1ak",
  "team_id": "team_n8q2x4m1ak",
  "status": "active",
  "query": "Track changes in https://example.com/products/widget-pro pricing and stock information",
  "frequency": "every day at 9am America/Los_Angeles",
  "cron_expression": "0 0 * * ? *",
  "notification_channel": "email",
  "notification_target": "alerts@example.com",
  "output_schema": {
    "type": "object",
    "properties": {
      "title": { "type": "string" },
      "published_at": { "type": "string" }
    },
    "required": ["title"]
  },
  "metadata": {
    "product_id": "widget-pro",
    "team": "growth"
  }
}

结构化监控器输出

在创建请求中设置 output_schema,当你希望监控器提取结果遵循特定的 JSON 结构时。该模式必须是有效的 JSON Schema。

通知渠道

监控器支持每个监控器一个渠道:
  • email: 设置一个 email 字段
  • webhook: 设置一个 webhook_url 字段
  • sms: 设置一个 phone_number 字段(E.164 格式,例如 +14155552671
每个请求中你只能设置其中一个字段。

Webhook 示例

{
  "query": "Watch for changes in https://example.com/terms legal terms",
  "frequency": "everyday at 10:00 am",
  "webhook_url": "https://hooks.example.com/olostep-monitor"
}

SMS 示例

{
  "query": "Watch for changes in https://example.com/terms legal terms",
  "frequency": "everyday at 10:00 am",
  "phone_number": "+14155552671"
}

时间频率

以自然语言设置 frequency,例如:
  • every day at 9am America/Los_Angeles
  • every 5 minutes
  • every 6 hours
API 会自动解释这些时间表文本并派生 cron 表达式。如果没有提供时区,默认使用 UTC。

列出监控器

使用 GET /v1/monitors 检索你团队的所有监控器。 默认情况下,已删除的监控器会被过滤掉。使用 ?include_deleted=true 来包含它们。
import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

response = requests.get(f"{API_URL}/monitors", headers=headers)
result = response.json()
print(f"Total monitors: {result['count']}")
print(json.dumps(result, indent=2))

响应结构

{
  "monitors": [
    {
      "id": "monitor_a3yk6z49p8",
      "object": "monitor",
      "created_at": 1777983215,
      "cron_expression": "13 13 * * ? *",
      "fda_id": "fda_a3yk6z49p8",
      "frequency": "daily",
      "metadata": {},
      "notification_channel": "email",
      "notification_target": "example@olostep.com",
      "query": "Gather the new blogposts from example website and send me updates with new posts.",
      "status": "active",
      "team_id": "team_a3yk6z49p8F",
      "updated_at": 1777983222
    }
  ],
  "count": 1
}

获取单个监控器

使用 GET /v1/monitors/:monitor_id 检索单个监控器。
import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"
MONITOR_ID = "monitor_n8q2x4m1ak"

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

response = requests.get(f"{API_URL}/monitors/{MONITOR_ID}", headers=headers)
print(json.dumps(response.json(), indent=2))

响应结构

{
  "id": "monitor_a3yk6z49p8",
  "object": "monitor",
  "created_at": 1777983215,
  "cron_expression": "13 13 * * ? *",
  "fda_id": "fda_a3yk6z49p8",
  "frequency": "daily",
  "metadata": {},
  "notification_channel": "email",
  "notification_target": "example@olostep.com",
  "query": "Gather the new blogposts from example website and send me updates with new posts.",
  "status": "active",
  "team_id": "team_a3yk6z49p8",
  "updated_at": 1777983222
}

列出监控器事件

使用 GET /v1/monitors/:monitor_id/events 列出监控器的快照事件。 此端点支持分页:
  • limit(默认 25,最大 100
  • cursor(来自先前响应的不透明分页令牌)
事件按最新优先返回。每个项目都包含一个短期有效的预签名 snapshot_url,以便你可以安全地获取私人快照内容。

示例

import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"
MONITOR_ID = "monitor_n8q2x4m1ak"

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

response = requests.get(
    f"{API_URL}/monitors/{MONITOR_ID}/events?limit=10",
    headers=headers
)
print(json.dumps(response.json(), indent=2))

响应结构

{
  "data": [
    {
      "id": "run_v7k2p9m3",
      "run_id": "run_v7k2p9m3",
      "created": 1777960800,
      "changed": true,
      "summary": "Price changed from $49 to $45 and stock moved to low availability.",
      "snapshot_url": "https://olostep-monitor-snapshot.s3.amazonaws.com/private/key?...signature..."
    }
  ],
  "has_more": false,
  "next_cursor": null
}
如果 cursor 格式错误,端点返回:
{
  "error": "Invalid cursor."
}

更新监控器

使用 POST /v1/monitors/:monitor_id 更新监控器。 支持的更新:
  • metadata(与现有元数据合并;传递空字符串值以删除键)
  • frequency(自然语言时间表文本,例如 every weekday at 08:30 America/New_York
如果你更新 frequency,API 会在内部重新创建监控器时间表。如果文本中未包含时区,则使用 UTC。

示例请求

import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"
MONITOR_ID = "monitor_n8q2x4m1ak"

payload = {
    "frequency": "every 2 hours",
    "metadata": {
        "owner": "ops-team",
        "deprecated_field": ""
    }
}

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

response = requests.post(
    f"{API_URL}/monitors/{MONITOR_ID}",
    headers=headers,
    json=payload
)
print(json.dumps(response.json(), indent=2))

暂停监控器

使用 POST /v1/monitors/:monitor_id/pause 暂停监控器。 暂停会禁用监控器的底层时间表,因此不会触发进一步的运行,并将监控器的 status 设置为 paused。监控器行、其配置及其过去的快照将被保留——只有未来计划的执行会停止。你可以稍后使用 POST /v1/monitors/:monitor_id/resume 恢复监控器。 只有 statusactive 的监控器可以暂停。请求体为空。
import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"
MONITOR_ID = "monitor_n8q2x4m1ak"

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

response = requests.post(f"{API_URL}/monitors/{MONITOR_ID}/pause", headers=headers)
print(response.status_code)
print(json.dumps(response.json(), indent=2))

响应

成功时,返回 200 和监控器,status 设置为 paused
{
  "id": "monitor_n8q2x4m1ak",
  "object": "monitor",
  "status": "paused",
  "updated_at": 1777986822,
  "query": "Track changes in https://example.com/products/widget-pro pricing and stock information",
  "frequency": "every day at 9am America/Los_Angeles",
  "notification_channel": "email",
  "notification_target": "alerts@example.com"
}
可能的错误响应:
  • 400monitor_id 格式错误,监控器已 deleted,监控器当前不 active,或监控器没有底层时间表。
  • 404 — 找不到监控器。
  • 409 — 在暂停尝试期间监控器状态发生变化(不再 active),或找不到底层时间表。

恢复监控器

使用 POST /v1/monitors/:monitor_id/resume 恢复暂停的监控器。 恢复会重新启用监控器的底层时间表,并将监控器的 status 设置回 active。计划的运行会在现有 frequency 上继续——不需要重新创建时间表。 只有 statuspaused 的监控器可以恢复。请求体为空。
import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"
MONITOR_ID = "monitor_n8q2x4m1ak"

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

response = requests.post(f"{API_URL}/monitors/{MONITOR_ID}/resume", headers=headers)
print(response.status_code)
print(json.dumps(response.json(), indent=2))

响应

成功时,返回 200 和监控器,status 设置回 active
{
  "id": "monitor_n8q2x4m1ak",
  "object": "monitor",
  "status": "active",
  "updated_at": 1777990422,
  "query": "Track changes in https://example.com/products/widget-pro pricing and stock information",
  "frequency": "every day at 9am America/Los_Angeles",
  "notification_channel": "email",
  "notification_target": "alerts@example.com"
}
可能的错误响应:
  • 400monitor_id 格式错误,监控器已 deleted,监控器当前不 paused,或监控器没有底层时间表。
  • 404 — 找不到监控器。
  • 409 — 在恢复尝试期间监控器状态发生变化(不再 paused),或找不到底层时间表。

删除监控器

使用 DELETE /v1/monitors/:monitor_id 删除监控器。 删除是软删除监控器行(status 变为 deleted),并且还会删除与该监控器相关的内部调度/影子代理资源。
import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"
MONITOR_ID = "monitor_n8q2x4m1ak"

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

response = requests.delete(f"{API_URL}/monitors/{MONITOR_ID}", headers=headers)
print(json.dumps(response.json(), indent=2))

示例用例

以下是 /v1/monitors 端点的实际示例请求。

跟踪 OpenAI API 的正常运行时间问题

运行每小时检查,监控 OpenAI API 状态,并在检测到正常运行时间问题时通过电子邮件通知你。
import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"

payload = {
    "query": "Track uptime issues in OpenAI API and email me at info@olostep.com",
    "frequency": "every hour",
    "email": "alerts@example.com"
}

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

response = requests.post(f"{API_URL}/monitors", headers=headers, json=payload)
print(response.status_code)
print(json.dumps(response.json(), indent=2))

跟踪产品评论——每次运行完整快照

监控产品页面,并在每次运行时将每条评论传送到 webhook。由于 query 没有要求与之前的运行去重,每次运行都会捕获页面上当前评论的完整快照(默认行为)。当你希望每次获取完整状态时使用此模式,例如覆盖下游表。
import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"

payload = {
    "query": "Monitor the reviews of this product https://www.walmart.com/reviews/product/957245477?sort=submission-desc and notify me by webhook. For each review extract the publication date, author, rating, title, and text content.",
    "frequency": "everyday at 9am",
    "webhook_url": "https://webhook.site/3fb2b00b-d66f-4c46-b778-f0c7b93d4d86",
    "output_schema": {
        "type": "object",
        "properties": {
            "reviews": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "publication_date": {"type": "string"},
                        "author": {"type": "string"},
                        "rating": {"type": "number"},
                        "title": {"type": "string"},
                        "text_content": {"type": "string"}
                    }
                },
                "required": [
                    "publication_date",
                    "author",
                    "rating",
                    "title",
                    "text_content"
                ]
            }
        }
    }
}

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

response = requests.post(f"{API_URL}/monitors", headers=headers, json=payload)
print(response.status_code)
print(json.dumps(response.json(), indent=2))

仅跟踪新租赁列表——与上次运行的增量

监控分页租赁搜索并运行多步骤管道,仅显示新出现的列表与上次运行相比,然后在通知 webhook 之前丰富和评分每个列表。完整的工作流——分页、检测新列表与上次检查点相比、用其位置的犯罪信号丰富每个新列表、根据价格/位置/安全性分配 1-10 的总体评分、持久化状态并通知——直接在 query 中表达为有序步骤。由于自然语言步骤描述了提取和增量逻辑,因此不需要 output_schema:仅 query 就能驱动获取、比较、丰富、评分和传递的内容。
import requests
import json

API_KEY = "<YOUR_API_KEY>"
API_URL = "https://api.olostep.com/v1"

payload = {
    "query": "Monitor the first 3 pages of https://streeteasy.com/for-rent/new-jersey/price:1000-1750?sort_by=listed_desc, detect only newly appeared listings versus the previous checkpoint, enrich each new listing with crime signals for its location, assign a 1-10 overall score based on price/location/safety, persist state, and notify via webhook.",
    "frequency": "every hour",
    "webhook_url": "https://webhook.site/3fb2b00b-d66f-4c46-b778-f0c7b93d4d86"
}

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

response = requests.post(f"{API_URL}/monitors", headers=headers, json=payload)
print(response.status_code)
print(json.dumps(response.json(), indent=2))

常见验证错误

监控器端点为常见的无效请求返回明确的验证错误:
  • 缺少 query
  • 缺少 frequency,或无效的时间表指令
  • 缺少所有的 emailwebhook_urlphone_number
  • 在同一请求中提供多个 emailwebhook_urlphone_number
  • 无效的电子邮件格式
  • 无效的 webhook URL(必须是 httphttps
  • 无效的电话号码格式(必须是 E.164,例如 +14155552671
  • 无效的 output_schema(必须是有效的 JSON Schema)
  • 无效的 monitor_id 格式
示例错误:
{
  "error": "Invalid frequency. Provide a natural-language schedule instruction."
}