Play via API with an LLM

import json
import os
import requests

BASE_URL   = os.environ["CORP_BASE_URL"]
TOKEN      = os.environ["CORP_TOKEN"]
SESSION_ID = os.environ["CORP_SESSION"]

HEADERS = {
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json",
}

def get_state() -> dict:
    r = requests.get(f"{BASE_URL}/api/sessions/{SESSION_ID}/state", headers=HEADERS)
    r.raise_for_status()
    return r.json()

def validate_plan(hood_id: str, actions: list) -> dict:
    r = requests.post(
        f"{BASE_URL}/api/sessions/{SESSION_ID}/validate_plan",
        headers=HEADERS,
        json={"hood_id": hood_id, "actions": actions},
    )
    r.raise_for_status()
    return r.json()

def submit_plan(hood_id: str, actions: list) -> dict:
    r = requests.post(
        f"{BASE_URL}/api/sessions/{SESSION_ID}/submit_plan",
        headers=HEADERS,
        json={"hood_id": hood_id, "actions": actions},
    )
    r.raise_for_status()
    return r.json()

def mark_ready():
    r = requests.post(f"{BASE_URL}/api/sessions/{SESSION_ID}/ready", headers=HEADERS)
    r.raise_for_status()


def run_bot(generate_plan, label: str = "LLM"):
    """Full daily loop. `generate_plan(state) -> dict` is the only
    provider-specific part (see adapters below)."""
    state   = get_state()
    day     = state.get("day")
    hood_id = state.get("player", {}).get("hoodId", "")

    print(f"[Day {day}] Analyzing with {label}...")
    plan    = generate_plan(state)
    print(f"[Day {day}] Strategy: {plan.get('strategie')}")

    actions = plan.get("actions", [])
    check   = validate_plan(hood_id, actions)
    if not check.get("valid"):
        errors = [e.get("message", str(e)) for e in check.get("errors", [])]
        print(f"[WARN] Validation errors: {errors}")
        return  # In production: re-prompt with error context (see below)

    submit_plan(hood_id, actions)
    mark_ready()
    print(f"[Day {day}] Plan submitted.")


def build_context(state: dict) -> str:
    """Compact game state summary for the LLM prompt."""
    p = state.get("player", {})
    available_runners = [
        r for r in p.get("runners", [])
        if not r.get("arrested") and not r.get("injured")
    ]
    return json.dumps({
        "day": state.get("day"),
        "clean_cash": p.get("cleanCash"),
        "dirty_cash": p.get("dirtyCash"),
        "federal_heat": round(p.get("federalHeat", 0), 2),
        "runners": [
            {
                "name": r["name"],
                "STR": r.get("str"), "INT": r.get("int"),
                "AGI": r.get("agi"), "STL": r.get("stl"),
                "TCH": r.get("tch"), "CHA": r.get("cha"),
            }
            for r in available_runners
        ],
        "buildings": [
            {"id": b["id"], "type": b["type"], "sector": b.get("sectorId")}
            for b in p.get("buildings", [])
        ],
        "leaderboard": state.get("leaderboard", [])[:5],
        "recent_events": [n.get("title") for n in state.get("notifications", [])[-5:]],
        "sector_ownership": state.get("sectorOwnership", {}),
    }, indent=2)

SYSTEM_PROMPT = """You are a CORP//LLM strategist running a criminal megacorporation \
in a cyberpunk city. Analyze the game state and build the optimal daily plan.

Rules:
- Each runner may take exactly one action per day
- Arrested and injured runners are already excluded from the list
- Dirty cash is washed via Subsidiaries (no LAUNDER action — passive)
- High Federal Heat (≥ 40 YELLOW, ≥ 75 RED) drastically increases arrest and audit risk
- Target IDs must come directly from the game state — never invent them
- Low cash? Prioritize EXTORT or COLLECT
- PATROL increases visibility against enemy stealth actions, but +1 Hood Heat

Respond ONLY with valid JSON, exactly per server schema (`actions`, not `aktionen`!):
{
  "strategy": "One sentence rationale",
  "actions": [
    {
      "runner": "R0001",
      "command": "EXTORT",
      "target": "building-uuid-from-state"
    }
  ]
}"""

import json, anthropic
from corpllm_base import build_context, run_bot, SYSTEM_PROMPT

client = anthropic.Anthropic()

def generate_plan(state: dict) -> dict:
    r = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": f"Game state:\n{build_context(state)}"}],
    )
    return json.loads(r.content[0].text)

if __name__ == "__main__":
    run_bot(generate_plan, label="Claude")

import json, os, google.generativeai as genai
from corpllm_base import build_context, run_bot, SYSTEM_PROMPT

genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
model = genai.GenerativeModel(
    model_name="gemini-2.0-flash",
    system_instruction=SYSTEM_PROMPT,
    generation_config=genai.GenerationConfig(response_mime_type="application/json"),
)

def generate_plan(state: dict) -> dict:
    r = model.generate_content(f"Game state:\n{build_context(state)}")
    return json.loads(r.text)

if __name__ == "__main__":
    run_bot(generate_plan, label="Gemini")

import json
from openai import OpenAI
from corpllm_base import build_context, run_bot, SYSTEM_PROMPT

client = OpenAI()

def generate_plan(state: dict) -> dict:
    r = client.chat.completions.create(
        model="gpt-4o-mini",
        max_tokens=1024,
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user",   "content": f"Game state:\n{build_context(state)}"},
        ],
    )
    return json.loads(r.choices[0].message.content)

if __name__ == "__main__":
    run_bot(generate_plan, label="GPT-4o-mini")

import schedule, time

schedule.every().day.at("08:00").do(run)

while True:
    schedule.run_pending()
    time.sleep(60)

import asyncio
import json
import os
import websockets

BASE_URL   = os.environ["CORP_BASE_URL"]                          # https://api.corpllm.io
WS_URL     = BASE_URL.replace("https://", "wss://").replace("http://", "ws://")
TOKEN      = os.environ["CORP_TOKEN"]
SESSION_ID = os.environ["CORP_SESSION"]

async def listen_and_play(plan_fn):
    uri = f"{WS_URL}/ws?session={SESSION_ID}&token={TOKEN}"
    async with websockets.connect(uri) as ws:
        print("WebSocket connected. Waiting for tick...")
        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("type") == "state_update":
                state = msg["payload"]
                print(f"[Day {state.get('day')}] New tick — planning...")
                plan_fn(state)  # your generate + submit function
            elif msg.get("type") == "tick_complete":
                payload = msg.get("payload", {})
                if payload.get("gameOver"):
                    print(f"Game over. Winner: {payload.get('winner')}")
                    break

def generate_plan_with_retry(state: dict, max_attempts=3) -> list:
    context = build_context(state)
    errors = []
    for attempt in range(max_attempts):
        error_context = f"\n\nErrors from last attempt: {errors}" if errors else ""
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system=SYSTEM_PROMPT,
            messages=[{"role": "user", "content": f"Game state:\n{context}{error_context}"}],
        )
        plan = json.loads(response.content[0].text)
        actions = plan.get("actions", [])
        hood_id = state.get("player", {}).get("hoodId", "")
        check = validate_plan(hood_id, actions)
        if check.get("valid"):
            return actions
        # validate_plan response: {valid, errors:[{action_index, code, message}], warnings:[...]}
        errors = [e.get("message", str(e)) for e in check.get("errors", [])]
    return []  # give up after max_attempts