from fastapi import FastAPI, HTTPException, Body
from fastapi.responses import HTMLResponse
from typing import Optional, Dict, Any
import uvicorn
import os
import sys
# Ensure current directory is in path for imports
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.append(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "email_triage_env"))
from email_triage_env.env import EmailTriageEnv
from email_triage_env.models import Action, Observation, Reward, State
from email_triage_env.tasks import TASKS
app = FastAPI(title="Email Triage OpenEnv")
env = EmailTriageEnv()
@app.get("/", response_class=HTMLResponse)
async def root():
state = env.state()
emails_html = ""
if state:
for email in state.all_emails:
status = "Archived" if email.is_archived else "Active"
emails_html += f"""
From: {email.sender}
Subject: {email.subject}
Status: {status}
{email.body}
"""
else:
emails_html = "Environment not reset yet. Call /reset to start.
"
return f"""
📧 Email Triage OpenEnv
Email Triage OpenEnv
Status: Running
Current Inbox State
{emails_html}
This is an OpenEnv server. Use the /step and /reset endpoints for programmatic interaction.
"""
@app.post("/reset", response_model=Observation)
async def reset(task_id: str = Body("easy", embed=True)):
try:
return env.reset(task_id)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/step")
async def step(action: Action):
try:
observation, reward, done, info = env.step(action)
return {
"observation": observation,
"reward": reward,
"done": done,
"info": info
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/state")
async def state():
try:
return env.state()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/tasks")
async def get_tasks():
return {
"tasks": [
{"id": task.id, "description": task.description, "difficulty": task.difficulty}
for task in TASKS.values()
]
}
def main():
uvicorn.run(app, host="0.0.0.0", port=7860)
if __name__ == "__main__":
main()