CrewAI ワークフロー Telemetry

import json
import os
import sys
from time import perf_counter
from urllib.request import Request, urlopen
from uuid import uuid4


def emit_crew_invocation(event):
    try:
        key = os.environ.get("TELEMETRY_API_KEY")
        if not key:
            return
        request = Request(
            "https://api.telemetry.sh/log",
            data=json.dumps({"table": "crew_invocation", "data": event}).encode("utf-8"),
            headers={"Authorization": key, "Content-Type": "application/json",
                     "User-Agent": "telemetry-crew-example/1"},
            method="POST",
        )
        with urlopen(request, timeout=1) as response:
            if not 200 <= response.status < 300:
                raise RuntimeError("telemetry_delivery_unconfirmed")
    except Exception:
        try:
            print("Telemetry delivery unconfirmed", file=sys.stderr)
        except Exception:
            pass


def run_research_crew(research_crew, inputs):
    # Use an already configured, synchronous, non-streaming Crew.
    if getattr(research_crew, "stream", False):
        raise ValueError("This wrapper requires stream=False")
    run_id = str(uuid4())
    started_at = perf_counter()
    status, error_type = "interrupted", None
    try:
        result = research_crew.kickoff(inputs=inputs)
        status = "returned"
        return result
    except Exception as error:
        status = "failed"
        error_type = "timeout" if isinstance(error, TimeoutError) else "crew_error"
        raise
    finally:
        # Neither metadata creation nor export may replace the crew outcome.
        try:
            emit_crew_invocation({
                "run_id": run_id,
                "crew_name": "research_crew",
                "status": status,
                "error_type": error_type,
                "duration_ms": round((perf_counter() - started_at) * 1000),
                "release": os.environ.get("APP_RELEASE"),
            })
        except Exception:
            pass


# Pass your application's existing Crew and inputs:
# result = run_research_crew(research_crew, inputs)