"""FastAPI server for the Feature Flag Cleanup OpenEnv environment."""
import sys
import os
# Add project root to path so env module can be imported
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from typing import Any, Dict, Optional
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse, HTMLResponse
from pydantic import BaseModel
from env.environment import FeatureFlagCleanupEnv
from env.models import FlagAction, FlagObservation, FlagState
app = FastAPI(
title="Feature Flag Cleanup Agent",
description="OpenEnv environment where AI agents learn to triage and clean up stale feature flags",
version="0.1.0",
)
# Global environment instance
_env = FeatureFlagCleanupEnv()
# --- Root endpoint ---
@app.get("/", response_class=HTMLResponse)
def root():
return """
Feature Flag Cleanup Agent
Feature Flag Cleanup Agent
Decide the fate of each feature flag: remove, keep, deprecate, or escalate
Step
-
Reward
-
Remaining
-
You're a senior engineer at a large tech company. Feature flags have piled up. For each flag, decide: should it be removed, kept, deprecated, or escalated for human review? You can also investigate to reveal hidden notes before deciding (costs -0.05).
Pick a difficulty above to begin.
Rollout
Action History
"""
# --- Request/Response Models ---
class ResetRequest(BaseModel):
task_id: str = "easy"
class ResetResponse(BaseModel):
observation: Dict[str, Any]
reward: Optional[float] = None
done: bool = False
class StepRequest(BaseModel):
action: Dict[str, Any]
timeout_s: Optional[float] = None
class StepResponse(BaseModel):
observation: Dict[str, Any]
reward: float
done: bool
info: Dict[str, Any] = {}
class StateResponse(BaseModel):
episode_id: Optional[str] = None
step_count: int = 0
task_id: str = "easy"
current_flag_index: int = 0
total_flags: int = 0
cumulative_reward: float = 0.0
flags_processed: list = []
class GradeResponse(BaseModel):
score: float
task_id: str
class SchemaResponse(BaseModel):
action: Dict[str, Any]
observation: Dict[str, Any]
state: Dict[str, Any]
class MetadataResponse(BaseModel):
name: str
description: str
version: str
author: str
# --- Endpoints ---
@app.get("/health")
def health():
return {"status": "healthy"}
@app.get("/metadata", response_model=MetadataResponse)
def metadata():
return MetadataResponse(
name="feature-flag-cleanup",
description="OpenEnv environment where AI agents learn to triage and clean up stale feature flags in a simulated engineering organization",
version="0.1.0",
author="Falguni",
)
@app.get("/schema", response_model=SchemaResponse)
def schema():
return SchemaResponse(
action=FlagAction.model_json_schema(),
observation=FlagObservation.model_json_schema(),
state=FlagState.model_json_schema(),
)
@app.post("/reset", response_model=ResetResponse)
def reset(request: ResetRequest = ResetRequest()):
obs = _env.reset(task_id=request.task_id)
return ResetResponse(observation=obs, reward=None, done=False)
@app.post("/step", response_model=StepResponse)
def step(request: StepRequest):
try:
obs, reward, done, info = _env.step(request.action)
return StepResponse(observation=obs, reward=reward, done=done, info=info)
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@app.get("/state", response_model=StateResponse)
def state():
s = _env.state()
return StateResponse(**s)
@app.post("/grade", response_model=GradeResponse)
def grade():
score = _env.grade()
s = _env.state()
return GradeResponse(score=score, task_id=s["task_id"])
@app.post("/mcp")
async def mcp(request: Request):
"""MCP (Model Context Protocol) JSON-RPC endpoint."""
body = await request.json()
# Return a valid JSON-RPC 2.0 response
return JSONResponse(content={
"jsonrpc": "2.0",
"id": body.get("id", 1),
"result": {
"name": "feature-flag-cleanup",
"description": "OpenEnv environment for feature flag triage and cleanup",
"tools": [
{
"name": "reset",
"description": "Reset the environment for a new episode",
"parameters": {"task_id": {"type": "string", "enum": ["easy", "medium", "hard"]}},
},
{
"name": "step",
"description": "Take an action on the current feature flag",
"parameters": {"action": {"type": "string", "enum": ["remove", "keep", "deprecate", "escalate"]}},
},
{
"name": "state",
"description": "Get current environment state",
"parameters": {},
},
],
},
})
def main():
"""Entry point for the server."""
import uvicorn
port = int(os.environ.get("PORT", 7860))
uvicorn.run("server.app:app", host="0.0.0.0", port=port, reload=False)
if __name__ == "__main__":
main()