Integrations

CrewAI

A CrewAI BaseTool that POSTs /v1/judge. The script also POSTs the documented consensus body.

All integrations

"""CrewAI tool that POSTs /v1/judge to Screwtape.ai."""
import json
import os
import urllib.request
from crewai import Agent
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

API = "https://screwtape.ai"


def screwtape_post(path: str, payload: dict) -> dict:
    """POST JSON to Screwtape.ai. Requires SCREWTAPE_API_KEY."""
    key = os.environ["SCREWTAPE_API_KEY"]
    request = urllib.request.Request(
        API + path,
        data=json.dumps(payload).encode(),
        headers={
            "Authorization": "Bearer " + key,
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(request) as response:
        body = json.load(response)
        if not isinstance(body, dict):
            raise RuntimeError("expected a JSON object")
        return body


JUDGE = {
    "task": "State the crew size and launch site.",
    "agent_output": "The crew size was 4. The mission launched from Cape Canaveral.",
    "evidence": [
        "The mission launched on 2024-03-01 from Cape Canaveral.",
        "The crew size was 4."
    ],
    "criteria": [
        "crew size",
        "Cape Canaveral"
    ]
}

CONSENSUS = {
    "task": "State the crew size.",
    "answers": [
        {
            "agent_id": "alpha",
            "output": "The crew size was 4."
        },
        {
            "agent_id": "beta",
            "output": "The crew size was 4."
        },
        {
            "agent_id": "gamma",
            "output": "The crew size was 4."
        }
    ]
}

class JudgeInput(BaseModel):
    task: str = Field(..., description="The task the other agent was given.")
    agent_output: str = Field(..., description="The output to judge.")


class ScrewtapeJudgeTool(BaseTool):
    name: str = "screwtape_judge"
    description: str = "Judge an agent output. Returns JSON with verdict, score, and reasons."
    args_schema: type[BaseModel] = JudgeInput

    def _run(self, task: str, agent_output: str) -> str:
        return json.dumps(screwtape_post("/v1/judge", {"task": task, "agent_output": agent_output}))


def reviewer() -> Agent:
    """Build a CrewAI agent that can call the judge tool. Requires your CrewAI model setup."""
    return Agent(
        role="Reviewer",
        goal="Judge another agent's answer before accepting it.",
        backstory="You call screwtape_judge and report the verdict, score, and reasons.",
        tools=[ScrewtapeJudgeTool()],
        verbose=False,
    )


if __name__ == "__main__":
    print(json.dumps(screwtape_post("/v1/judge", JUDGE)))
    print(json.dumps(screwtape_post("/v1/consensus", CONSENSUS)))