Spaces:
Sleeping
Sleeping
File size: 2,650 Bytes
6a16b0d 64c0edc 6a16b0d afd7056 64c0edc afd7056 6a16b0d 6fbd11f 6a16b0d afd7056 6a16b0d 64c0edc 2951c44 64c0edc 6a16b0d 64c0edc 6a16b0d 64c0edc afd7056 64c0edc 2951c44 64c0edc 2951c44 64c0edc 6a16b0d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 |
from fastapi import FastAPI, Body, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List, Dict, Any
from pathlib import Path
from datetime import datetime
import subprocess
import uuid
import pandas as pd
app = FastAPI(title="Agent Sandbox API")
class IncommingMail(BaseModel):
data: List[Dict[str, Any]] # requires a "data" field in JSON
timestamp: str = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
id: str = uuid.uuid4().hex[:5]
@property
def file_path(self):
return f"{self.timestamp}_{self.id}_sample.parquet"
def store_mail(self):
try:
import pandas as pd
data = pd.DataFrame(self.data)
data.to_parquet(self.file_path)
except Exception as e:
raise e
class MailResponse(BaseModel):
status: str
timestamp: str
pigeon_hole: str
pigeon_location: str
class CodeScript(BaseModel):
code: str
@app.post("/run")
def run_code(payload: CodeScript):
"""
Execute Python code in a sandboxed subprocess.
Input JSON: {"code": "print(2+2)"} or {"code": "2+2"}
Output JSON: {"output": "4"} or {"error": "..."}
"""
code = payload.code
if not code:
raise HTTPException(status_code=400, detail="Received Payload in incorrect format. Must pass key: 'code' with value of the script to run python code.")
try:
# Run the code in a subprocess with a 5s timeout
proc = subprocess.run(
["python3", "-c", code],
capture_output=True,
text=True,
timeout=5
)
if proc.returncode != 0:
return {"error": proc.stderr.strip() or "Unknown error"}
output = proc.stdout.strip()
# If nothing printed, try evaluating expression directly
if not output:
try:
result = eval(code, {})
output = str(result)
except Exception:
output = "<no output>"
return {"output": output}
except subprocess.TimeoutExpired:
raise HTTPException(status_code=408, detail="Execution timed out")
@app.post("/inbox", response_model=MailResponse)
def agent_mailbox(mail: IncommingMail):
"""
Save incoming data with a timestamp + unique ID.
"""
# Prepare / fetch directory
inbox_dir = Path("inbox")
inbox_dir.mkdir(exist_ok=True)
mail.store_mail()
return MailResponse(
status="Data Saved!",
timestamp=mail.timestamp,
pigeon_hole=mail.id,
pigeon_location=mail.file_path
)
|