PingPulse AI Agent Integration Guide

export PINGPULSE_API_KEY="ppk_your_key_here"
export PINGPULSE_WORKFLOW_ID="dwf_your_workflow_id"

curl -X POST "https://pingpulsehq.com/dhk/my-workflow-id?key=${PINGPULSE_API_KEY}&start=yes&stagePath=1" \
  -H "Content-Type: application/json" \
  -d '{"stagename": "planning", "agent": "planner-agent"}'

curl -X POST "https://pingpulsehq.com/dhk/${PINGPULSE_WORKFLOW_ID}?key=${PINGPULSE_API_KEY}&stagePath=2" \
  -H "Content-Type: application/json" \
  -d '{"stagename": "research", "agent": "researcher"}'

curl -X POST "https://pingpulsehq.com/dhk/$WORKFLOW_ID?stagePath=3.1-1" \
  -H "Content-Type: application/json" \
  -d '{
    "stagename": "Code Analysis",
    "agent": "analyzer",
    "status": "completed",
    "key": "ppk_xxx_yyy",
    "model_id": "claude-3-opus-20240229",
    "input_tokens": 15420,
    "output_tokens": 3847,
    "cost_usd": 0.42,
    "tool_calls": 5,
    "latency_ms": 8432
  }'

def ping_with_llm_metrics(
    stage: str,
    agent: str,
    model_id: str = None,
    input_tokens: int = None,
    output_tokens: int = None,
    cost_usd: float = None,
    tool_calls: int = None,
    **kwargs
):
    """Send a ping with optional LLM cost metrics."""
    data = {
        "stage": stage,
        "agent": agent,
        "status": "completed"
    }

    # Add optional LLM metrics
    if model_id:
        data["model_id"] = model_id
    if input_tokens is not None:
        data["input_tokens"] = input_tokens
    if output_tokens is not None:
        data["output_tokens"] = output_tokens
    if cost_usd is not None:
        data["cost_usd"] = cost_usd
    if tool_calls is not None:
        data["tool_calls"] = tool_calls

    response = requests.post(
        f"{PINGPULSE_BASE_URL}/dhk/{WORKFLOW_ID}",
        params={"key": API_KEY},
        json=data
    )
    return response.json()

# Usage after an LLM call
result = ping_with_llm_metrics(
    stage="3.1",
    agent="coder",
    model_id="gpt-4-turbo",
    input_tokens=usage.prompt_tokens,
    output_tokens=usage.completion_tokens,
    cost_usd=calculate_cost(usage),
    tool_calls=len(response.tool_calls) if response.tool_calls else 0
)

curl -X POST "https://pingpulsehq.com/api/v1/agents" \
  -H "Content-Type: application/json" \
  -d '{
    "key": "'${PINGPULSE_API_KEY}'",
    "code": "my-agent",
    "name": "My AI Agent",
    "description": "Handles research tasks",
    "color": "#6366f1"
  }'

curl -X POST "https://pingpulsehq.com/api/approvals" \
  -H "Content-Type: application/json" \
  -d '{
    "key": "'${PINGPULSE_API_KEY}'",
    "context": "Deploy version 2.3.1 to production? This will affect 10,000 users.",
    "timeout_seconds": 300,
    "timeout_action": "deny"
  }'

{
  "approval_id": "apr_a1b2c3d4e5f6",
  "status": "pending",
  "result": null,
  "context": "Deploy version 2.3.1 to production?...",
  "timeout_seconds": 300,
  "seconds_remaining": 299,
  "expires_at": "2025-01-15T10:05:00Z"
}

curl "https://pingpulsehq.com/api/approvals/apr_a1b2c3d4e5f6?key=${PINGPULSE_API_KEY}"

import os
import time
import requests

API_KEY = os.getenv("PINGPULSE_API_KEY")
BASE_URL = "https://pingpulsehq.com"

def request_approval(context: str, timeout_seconds: int = 300) -> dict:
    """Request human approval and wait for response."""
    # Step 1: Create approval request
    response = requests.post(
        f"{BASE_URL}/api/approvals",
        json={
            "key": API_KEY,
            "context": context,
            "timeout_seconds": timeout_seconds,
            "timeout_action": "deny"
        }
    )
    approval = response.json()
    approval_id = approval["approval_id"]
    print(f"Approval requested: {approval_id}")
    print(f"Waiting for human response (timeout: {timeout_seconds}s)...")

    # Step 2: Poll for result
    while True:
        response = requests.get(
            f"{BASE_URL}/api/approvals/{approval_id}",
            params={"key": API_KEY}
        )
        result = response.json()

        if result["status"] != "pending":
            return result

        # Still pending - wait and poll again
        remaining = result.get("seconds_remaining", 0)
        print(f"  Waiting... ({remaining}s remaining)")
        time.sleep(min(5, remaining))  # Poll every 5 seconds

# Usage
approval = request_approval(
    context="Delete 500 old records from the database?",
    timeout_seconds=120
)

if approval["result"] == 1:
    print("APPROVED - proceeding with deletion")
    # do_deletion()
else:
    print(f"DENIED - reason: {approval.get('notes', 'No reason given')}")
    # abort_operation()

{
  "status": "success",
  "message": "Stage 'research' recorded",
  "instance_id": "d_abc12345",
  "stage_execution_id": 42
}