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
|
from typing import List
from ..registry import action
@action(
name="list_files",
description="List files in a directory",
parameters=[
{
"name": "path",
"description": "Path to the directory",
"type": "string",
"required": True,
}
],
output_type="list[str]",
)
async def list_files(agent, task_id: str, path: str) -> List[str]:
"""
List files in a workspace directory
"""
return agent.workspace.list(task_id=task_id, path=str(path))
@action(
name="write_file",
description="Write data to a file",
parameters=[
{
"name": "file_path",
"description": "Path to the file",
"type": "string",
"required": True,
},
{
"name": "data",
"description": "Data to write to the file",
"type": "bytes",
"required": True,
},
],
output_type="None",
)
async def write_file(agent, task_id: str, file_path: str, data: bytes):
"""
Write data to a file
"""
if isinstance(data, str):
data = data.encode()
agent.workspace.write(task_id=task_id, path=file_path, data=data)
return await agent.db.create_artifact(
task_id=task_id,
file_name=file_path.split("/")[-1],
relative_path=file_path,
agent_created=True,
)
@action(
name="read_file",
description="Read data from a file",
parameters=[
{
"name": "file_path",
"description": "Path to the file",
"type": "string",
"required": True,
},
],
output_type="bytes",
)
async def read_file(agent, task_id: str, file_path: str) -> bytes:
"""
Read data from a file
"""
return agent.workspace.read(task_id=task_id, path=file_path)
|
