Events API

GET /v1/requests?q=<search>&from=<ISO time>&to=<ISO time>&limit=50&cursor=<next_cursor>

import os
 
import httpx
 
API = os.environ.get("CLI_BASE_URL", "https://cci.gitdate.ink/api/v1").rstrip("/")
http = httpx.Client(base_url=API + "/", headers={"Authorization": f"Bearer {os.environ['CLI_API_KEY']}"},
                    timeout=30.0)
 
 
def requests_matching(q: str, **params):
    cursor = None
    while True:
        page = http.get("requests", params={"q": q, "limit": 500, "cursor": cursor, **params}).json()
        yield from page["requests"]
        cursor = page["next_cursor"]
        if cursor is None:
            return
 
 
for row in requests_matching("operation:evaluate outcome:fail_closed profile:support-routing-v3",
                             **{"from": "2026-09-01T00:00:00Z"}):
    print(row["request_id"], row["status"], row["duration_ms"], row["heuristic_count"])

op:evaluate outcome:fail_closed profile:triage-set-v3 duration_ms:>1000
provider:openai -status:200 env:production
heuristic:true tag:pilot meta.release:"2026.10.1"
auth:revoked

GET /v1/requests/{request_id}?include_payload=true

{
  "request": {"request_id": "req_7d1e4a90c2b35f18", "status": 424, "outcome": "fail_closed", "...": "..."},
  "spans": [{"span_id": "...", "parent_span_id": "...", "name": "profile.resolve triage-set-v3", "...": "..."}],
  "answers": [{"query_id": "department", "query_type": "set", "guarantee_type": "heuristic", "...": "..."}],
  "payload": {"request": {"...": "..."}, "response": {"...": "..."}, "redactions": 2, "truncated": false},
  "guard_events": [],
  "attempts": [{"request_id": "req_...", "status": 502, "outcome": "backend_error", "replay_of": null}],
  "request_id": "req_..."
}

GET /v1/usage?group_by=key&from=<ISO time>&to=<ISO time>

usage = http.get("usage", params={"group_by": "key", "from": "2026-09-01T00:00:00Z"}).json()
for row in usage["usage"]:
    print(row["group"], row["requests"], row["cost_usd"], row["p95_ms"])

GET /v1/calibration-profiles/{name}/events?limit=200

POST /v1/events
{"events": [ ... up to 500 ... ]}

# cci_events.py
import contextvars
import os
import uuid
from datetime import datetime, timezone
from typing import Any, Iterable, Mapping, Optional
 
import httpx
 
BASE_URL = os.environ.get("CLI_BASE_URL", "https://cci.gitdate.ink/api/v1").rstrip("/") + "/"
_http = httpx.Client(base_url=BASE_URL, timeout=10.0,
                     headers={"Authorization": f"Bearer {os.environ['CLI_API_KEY']}"})
 
# The current thread, set by your code before each agent run.
current_thread: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar("cci_thread", default=None)
 
 
def event(type: str, *, event_id: Optional[str] = None, **fields: Any) -> dict[str, Any]:
    """One guard event; fields that are None are left out."""
    out: dict[str, Any] = {"type": type, "event_id": event_id or str(uuid.uuid4()),
                           "at": datetime.now(timezone.utc).isoformat()}
    out.update({key: value for key, value in fields.items() if value is not None})
    return out
 
 
def decision_event(decision: Mapping[str, Any], **fields: Any) -> dict[str, Any]:
    """A guard.decision event from GuardDecision.to_dict(), or a decision dict an integration stored."""
    guarantee = decision.get("guarantee") or {}
    return event("guard.decision", tool_name=decision.get("tool"), action=decision.get("action"),
                 profile=guarantee.get("calibration_profile"),
                 attributes={"reason": decision.get("reason"), "guarantee": guarantee.get("statement"),
                             "arguments": decision.get("arguments")},
                 **fields)
 
 
def send(events: Iterable[Mapping[str, Any]]) -> Optional[dict]:
    """Post a batch of at most 500 events. Best effort: returns None on any failure."""
    batch = list(events)
    if not batch:
        return None
    try:
        response = _http.post("events", json={"events": batch})
        response.raise_for_status()
        return response.json()
    except httpx.HTTPError:
        return None
 
 
def _tag(request: httpx.Request) -> None:
    thread = current_thread.get()
    if thread:
        request.headers["X-CCI-Thread-Id"] = thread
 
 
def traced_http_client(timeout: float = 60.0) -> httpx.Client:
    """For CLIClient(http_client=...): every evaluate carries the current thread id."""
    return httpx.Client(timeout=timeout, event_hooks={"request": [_tag]})

from cli_sdk import CLIClient
from cci_events import traced_http_client
 
client = CLIClient(http_client=traced_http_client())

from cli_sdk.integrations import ToolGuard
from cci_events import current_thread, decision_event, send
 
 
class ReportingToolGuard(ToolGuard):
    """A ToolGuard that also reports each decision as a guard.decision event."""
 
    def __init__(self, *args, framework, agent_name=None, **kwargs):
        super().__init__(*args, **kwargs)
        self.framework, self.agent_name = framework, agent_name
 
    def check(self, tool, arguments, state=None):
        decision = super().check(tool, arguments, state)
        send([decision_event(decision.to_dict(), framework=self.framework, agent_name=self.agent_name,
                             thread_id=current_thread.get())])
        return decision

from cci_events import current_thread, decision_event, event, send
 
 
def report_langgraph(graph, config, *, agent_name=None) -> None:
    thread = config["configurable"]["thread_id"]
    record = graph.get_state(config).values.get("cli_guard") or {}
    events = []
    for call in record.get("calls", []):
        events.append(decision_event(call, framework="langgraph", agent_name=agent_name, thread_id=thread,
                                     tool_call_id=call.get("id"), event_id=f"{thread}:{call.get('id')}:decision"))
    review = record.get("review")
    if review:
        for call in record.get("calls", []):
            events.append(event("guard.review", event_id=f"{thread}:{call.get('id')}:review",
                                framework="langgraph", agent_name=agent_name, thread_id=thread,
                                tool_call_id=call.get("id"), tool_name=call.get("tool"),
                                outcome=review.get("outcome"), attributes={"note": review.get("note")}))
    send(events)
 
 
config = {"configurable": {"thread_id": "th_77"}}
current_thread.set("th_77")
result = graph.invoke({"messages": [("user", claim_text)]}, config)
report_langgraph(graph, config, agent_name="claims-triage")

from typing import Annotated
 
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import InjectedToolCallId, tool
 
from cci_events import event, send
 
 
@tool
def pay_claim(claim_id: str, amount: float, config: RunnableConfig,
              tool_call_id: Annotated[str, InjectedToolCallId]) -> str:
    """Pay an approved claim."""
    receipt = payments.pay(claim_id, amount)
    send([event("guard.executed", event_id=f"{tool_call_id}:executed", framework="langgraph",
                thread_id=config["configurable"]["thread_id"], tool_call_id=tool_call_id, tool_name="pay_claim")])
    return receipt

from cli_sdk.integrations.langchain import guard_decisions
from cci_events import decision_event, event, send
 
 
def report_langchain(result, config, *, agent_name=None) -> None:
    thread = config["configurable"]["thread_id"]
    send(decision_event(decision, framework="langchain", agent_name=agent_name, thread_id=thread,
                        tool_call_id=call_id, event_id=f"{thread}:{call_id}:decision")
         for call_id, decision in guard_decisions(result).items())
 
 
def report_langchain_reviews(pending, decisions, config, *, reviewer=None) -> None:
    """`pending` is pending_reviews(result); `decisions` is the list passed to Command(resume=...)."""
    thread = config["configurable"]["thread_id"]
    send(event("guard.review", event_id=f"{thread}:{item.get('interrupt_id')}:{i}:review", framework="langchain",
               thread_id=thread, tool_name=item.get("name"), outcome=decision.get("type"), reviewer=reviewer)
         for i, (item, decision) in enumerate(zip(pending, decisions)))

from cci_events import decision_event, event, send
 
 
def report_adk(session, *, agent_name=None) -> None:
    events = []
    for i, entry in enumerate(session.state.get("cli_guard_audit", [])):
        ids = {"framework": "google-adk", "agent_name": agent_name, "session_id": session.id,
               "thread_id": session.id, "event_id": f"{session.id}:{i}"}
        if entry.get("action") in ("human_approved", "human_rejected"):
            outcome = "approve" if entry["action"] == "human_approved" else "reject"
            events.append(event("guard.review", tool_name=entry.get("tool"), outcome=outcome, **ids))
        else:
            events.append(decision_event(entry, **ids))
    send(events)

from cli_sdk.integrations.agent_framework import pending_reviews, review_message
from cci_events import current_thread, event, send
 
 
async def answer_review(agent, session, result, approved: bool, reviewer: str):
    review = pending_reviews(result)[0]
    request_id = getattr(review["request"], "id", None)
    send([event("guard.review", event_id=f"{request_id}:review" if request_id else None, framework="agent-framework",
                thread_id=current_thread.get(), tool_name=review["tool"],
                outcome="approve" if approved else "reject", reviewer=reviewer)])
    return await agent.run(review_message(review, approved), session=session)