from typing import Optional from fastapi import FastAPI from fastapi.responses import JSONResponse, HTMLResponse from pydantic import BaseModel try: from SOC_env.server.SOC_env_environment import SOCEnvironment from SOC_env.models import SOCAction, SOCObservation from SOC_env.tasks import TASKS as TASK_REGISTRY except ModuleNotFoundError: from server.SOC_env_environment import SOCEnvironment from models import SOCAction, SOCObservation from tasks import TASKS as TASK_REGISTRY app = FastAPI(title="SOC Incident Response Environment", version="0.1.0") _env = SOCEnvironment() class ResetRequest(BaseModel): task: Optional[str] = None difficulty: Optional[str] = "random" class StepRequest(BaseModel): decision: str reasoning: Optional[str] = None @app.get("/health") async def health(): return {"status": "healthy"} @app.get("/metadata") async def metadata(): return { "name": "SOC_env", "description": "SOC Incident Response Environment — AI agent acts as a Tier-1 SOC analyst triaging security alerts across easy, medium, and hard scenarios.", "version": "0.1.0", "author": "ApoorvaBadoni", } @app.post("/reset") async def reset(req: ResetRequest = ResetRequest()): global _env pinned_id = None difficulty = req.difficulty or "random" if req.task and req.task in TASK_REGISTRY: task_def = TASK_REGISTRY[req.task] pinned_id = task_def["scenario_id"] difficulty = task_def["difficulty"] _env = SOCEnvironment(difficulty=difficulty, pinned_scenario_id=pinned_id) obs = _env.reset() return JSONResponse({"observation": obs.model_dump(), "reward": 0.0, "done": False}) @app.post("/step") async def step(req: StepRequest): action = SOCAction(decision=req.decision, reasoning=req.reasoning) obs = _env.step(action) return JSONResponse({"observation": obs.model_dump(), "reward": obs.reward, "done": obs.done}) @app.get("/state") async def state(): s = _env.state return JSONResponse({"episode_id": s.episode_id, "step_count": s.step_count}) @app.get("/tasks") async def list_tasks(): return JSONResponse({"tasks": [ {"name": t["name"], "display_name": t["display_name"], "difficulty": t["difficulty"], "description": t["description"], "max_steps": t["max_steps"], "success_criteria": t["success_criteria"]} for t in TASK_REGISTRY.values() ]}) @app.get("/schema") async def schema(): from openenv.core.env_server.types import State return JSONResponse({ "action": SOCAction.model_json_schema(), "observation": SOCObservation.model_json_schema(), "state": State.model_json_schema(), }) @app.post("/mcp") async def mcp(): return JSONResponse({ "jsonrpc": "2.0", "id": None, "result": { "name": "SOC_env", "description": "SOC Incident Response Environment", "tools": [] } }) @app.get("/web", response_class=HTMLResponse) @app.get("/", response_class=HTMLResponse) async def web_ui(): return HTMLResponse(content=WEB_UI) def main(host: str = "0.0.0.0", port: int = 8000): import uvicorn uvicorn.run(app, host=host, port=port, reload=False) if __name__ == "__main__": main() WEB_UI = """ SOC Incident Response Environment
🔐

SOC Incident Response Environment

OpenEnv · RL Training Environment · Tier-1 SOC Analyst Simulation
LIVE

Select Task

Loading tasks...

Active Alert

Select a task to begin incident response

Response Actions

🔍 Investigation
🛡️ Containment
📋 Resolution

Episode Complete

Select a task above to start a new episode

Episode Stats

0
Step
0.00
Score
—
Max Steps

Analyst Feedback

detection

Investigation Context

Use investigate, query_logs, check_threat_intel, or run_sandbox to reveal context.

Episode Log

"""