main.py

"""CodeRifts enforcement with OpenAI Function Calling — abort on BLOCK."""
import json
import requests

CODERIFTS_URL = "https://app.coderifts.com/api/v1/public/preflight"


def coderifts_preflight(spec_url: str) -> dict:
    """Call CodeRifts public preflight."""
    resp = requests.get(f"{CODERIFTS_URL}?url={spec_url}")
    return resp.json()


# ── OpenAI function definitions ──────────────────────────────────────────

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "call_external_api",
            "description": "Call an external API endpoint",
            "parameters": {
                "type": "object",
                "properties": {
                    "spec_url": {
                        "type": "string",
                        "description": "The OpenAPI spec URL for the API",
                    },
                    "endpoint": {
                        "type": "string",
                        "description": "The endpoint path to call",
                    },
                },
                "required": ["spec_url", "endpoint"],
            },
        },
    }
]


def handle_tool_call(tool_call) -> str:
    """Process a tool call with CodeRifts enforcement."""
    args = json.loads(tool_call["arguments"])
    spec_url = args.get("spec_url", "")

    # CodeRifts guard — check before executing
    decision = coderifts_preflight(spec_url)
    print(f"[CODERIFTS] decision={decision['decision']} risk={decision.get('risk_score')}")

    if decision["decision"] == "BLOCK":
        return json.dumps({
            "error": "blocked_by_coderifts",
            "decision": "BLOCK",
            "risk_score": decision.get("risk_score"),
            "reason": "API spec failed safety check",
        })

    if decision["decision"] == "PENDING":
        return json.dumps({
            "status": "pending",
            "message": "Safety check in progress, retry in 5 seconds",
        })

    # Safe to proceed
    print(f"[OPENAI] Executing {args.get('endpoint')} on {spec_url}")
    return json.dumps({"status": "success", "data": {"order_id": "12345"}})


# ── Simulation ───────────────────────────────────────────────────────────
# In production, use:
#   from openai import OpenAI
#   client = OpenAI()
#   response = client.chat.completions.create(model="gpt-4", tools=TOOLS, ...)
#   for tool_call in response.choices[0].message.tool_calls:
#       result = handle_tool_call(tool_call)

def run_openai_demo(spec_url: str):
    """Simulate an OpenAI function calling loop with CodeRifts."""
    print(f"[OPENAI] Model wants to call API at {spec_url}")

    # Simulated tool call from the model
    tool_call = {
        "name": "call_external_api",
        "arguments": json.dumps({"spec_url": spec_url, "endpoint": "/orders/123"}),
    }

    result = handle_tool_call(tool_call)
    parsed = json.loads(result)

    if "error" in parsed:
        print(f"[OPENAI] Tool call blocked: {parsed['reason']}")
    else:
        print(f"[OPENAI] Tool call succeeded: {parsed}")


if __name__ == "__main__":
    run_openai_demo("https://petstore3.swagger.io/api/v3/openapi.json")