-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathtoolkit.py
151 lines (135 loc) Β· 5.5 KB
/
toolkit.py
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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
import json
from typing import Callable, Optional
from hide.client.hide_client import HideClient
from hide.model import OverwriteUpdate, Project, UdiffUpdate
class Toolkit:
def __init__(self, project: Project, client: HideClient) -> None:
self.project = project
self.client = client
def get_tasks(self) -> str:
"""Get the available tasks and their aliases in the project."""
try:
tasks = self.client.get_tasks(self.project.id)
return json.dumps([task.model_dump() for task in tasks])
except Exception as e:
return f"Failed to get tasks: {e}"
def run_task(
self,
command: Optional[str] = None,
alias: Optional[str] = None,
timeout: Optional[int] = None,
) -> str:
"""
Run a task in the project. Provide either command or alias. Set timeout in seconds. Command will be executed in the shell.
For the list of available tasks and their aliases, use the `get_tasks` tool.
"""
try:
result = self.client.run_task(
project_id=self.project.id, command=command, alias=alias
)
return f"exit code: {result.exit_code}\nstdout: {result.stdout}\nstderr: {result.stderr}"
except Exception as e:
return f"Failed to run task: {e}"
def create_file(self, path: str, content: str) -> str:
"""Create a file in the project."""
try:
file = self.client.create_file(
project_id=self.project.id, path=path, content=content
)
return f"File created:\n{file}"
except Exception as e:
return f"Failed to create file: {e}"
def apply_patch(self, path: str, patch: str) -> str:
"""Apply a patch to a file in the project. Patch must be in the unified diff format."""
try:
file = self.client.update_file(
project_id=self.project.id,
path=path,
update=UdiffUpdate(patch=patch),
)
return f"File updated:\n{file}"
except Exception as e:
return f"Failed to apply patch: {e}"
def insert_lines(self, path: str, start_line: int, content: str) -> str:
"""Insert lines in a project file. Lines are 1-indexed."""
try:
file = self.client.get_file(project_id=self.project.id, path=path)
file = file.insert_lines(start_line, content)
file = self.client.update_file(
project_id=self.project.id,
path=file.path,
update=OverwriteUpdate(content=file.content()),
)
return f"File updated:\n{file}"
except Exception as e:
return f"Failed to insert lines: {e}"
def replace_lines(
self, path: str, start_line: int, end_line: int, content: str
) -> str:
"""
Replace lines in a project file. Lines are 1-indexed.
start_line is inclusive. end_line is exclusive.
"""
try:
file = self.client.get_file(project_id=self.project.id, path=path)
file = file.replace_lines(start_line, end_line, content)
file = self.client.update_file(
project_id=self.project.id,
path=file.path,
update=OverwriteUpdate(content=file.content()),
)
return f"File updated:\n{file}"
except Exception as e:
return f"Failed to replace lines: {e}"
def append_lines(self, path: str, content: str) -> str:
"""Append lines to a file in the project."""
try:
file = self.client.get_file(project_id=self.project.id, path=path)
file = file.append_lines(content)
file = self.client.update_file(
project_id=self.project.id,
path=file.path,
update=OverwriteUpdate(content=file.content()),
)
return f"File updated:\n{file}"
except Exception as e:
return f"Failed to append lines: {e}"
def get_file(self, path: str) -> str:
"""Get a file from the project."""
try:
file = self.client.get_file(project_id=self.project.id, path=path)
return f"{file}"
except Exception as e:
return f"Failed to get file: {e}"
def delete_file(self, path: str) -> str:
"""Delete a file from the project."""
try:
deleted = self.client.delete_file(project_id=self.project.id, file=path)
return (
f"File deleted: {path}" if deleted else f"Failed to delete file: {path}"
)
except Exception as e:
return f"Failed to delete file: {e}"
def list_files(self) -> str:
"""List files in the project."""
try:
files = self.client.list_files(project_id=self.project.id)
return "\n".join([file.path for file in files])
except Exception as e:
return f"Failed to list files: {e}"
def get_tools(self) -> list[Callable[..., str]]:
return [
self.append_lines,
self.apply_patch,
self.create_file,
self.delete_file,
self.get_file,
self.get_tasks,
self.insert_lines,
self.list_files,
self.replace_lines,
self.run_task,
]
def as_langchain(self) -> "LangchainToolkit":
from hide.langchain.toolkit import LangchainToolkit
return LangchainToolkit(toolkit=self)