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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
import io
import json
import logging
import shutil
from pathlib import Path
from random import randint
from typing import Annotated, Any, Dict, List
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
artifacts: List[Dict[str, Any]] = []
class Task(BaseModel):
input: str
@app.post("/agent/tasks/{task_id}/artifacts")
async def upload_file(
task_id: str, file: Annotated[UploadFile, File()], relative_path: str = Form("")
) -> Dict[str, Any]:
logger.info(
"Uploading file for task_id: %s with relative path: %s", task_id, relative_path
)
absolute_directory_path = Path(__file__).parent.absolute()
save_path = (
absolute_directory_path
/ "agent/gpt-engineer"
/ "projects/my-new-project/workspace"
)
random_string = str(randint(0, 100000))
while random_string in artifacts:
random_string = str(randint(0, 100000))
artifact_data = await file.read()
artifacts.append(
{
"binary": artifact_data,
"relative_path": relative_path,
"file_name": file.filename,
"artifact_id": random_string,
}
)
print(artifacts)
return {
"artifact_id": random_string,
"file_name": "file_name",
"relative_path": "relative_path",
}
@app.get("/agent/tasks/{task_id}/artifacts")
async def get_files() -> List[Dict[str, Any]]:
logger.info("Fetching list of files for task")
return artifacts
@app.get("/agent/tasks/{task_id}/artifacts/{artifact_id}")
async def get_file(artifact_id: str):
for artifact in artifacts:
if artifact["artifact_id"] == artifact_id:
break
else:
logger.error("Attempt to access nonexistent artifact with ID: %s", artifact_id)
raise HTTPException(status_code=404, detail="Artifact not found")
logger.info("Fetching artifact with ID: %s", artifact_id)
# find aritifact where artifact_id = artifact_id
for artifact in artifacts:
if artifact["artifact_id"] == artifact_id:
return StreamingResponse(
io.BytesIO(artifact["binary"]),
media_type="application/octet-stream",
headers={"Content-Disposition": f"attachment; filename=test.txt"},
)
# return 404
return HTTPException(status_code=404, detail="Artifact not found")
@app.post("/agent/tasks/{task_id}/steps")
async def create_steps(task_id: str):
logger.info("Creating step for task_id: %s", task_id)
return {
"input": "random",
"additional_input": {},
"task_id": task_id,
"step_id": "random_step",
"name": "random",
"status": "created",
"output": "random",
"additional_output": {},
"artifacts": [],
"is_last": True,
}
@app.post("/agent/tasks")
async def create_tasks(task: Task):
artifacts.clear()
return {
"input": "random",
"additional_input": {},
"task_id": "static_task_id",
"artifacts": [],
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
|
