Skip to content

Tools

pycodeloop.tools.filesystem.ReadFileTool

Bases: Tool

Source code in pycodeloop/tools/filesystem.py 
 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
class ReadFileTool(Tool):
    name = "read_file"
    description = (
        "Read a text file's contents, optionally a line range. If you "
        "already read the exact same path and range in this session and "
        "it hasn't changed on disk since, this returns a short notice "
        "instead of repeating the content, to save tokens — pass "
        "force=true to get the full content again (e.g. if it scrolled "
        "out of your context after compaction)."
    )
    parameters = {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "File path to read"},
            "offset": {
                "type": "integer",
                "description": "1-indexed start line",
            },
            "limit": {"type": "integer", "description": "Max lines to read"},
            "force": {
                "type": "boolean",
                "description": "Show full content even if unchanged since your last read.",
                "default": False,
            },
        },
        "required": ["path"],
    }

    def __init__(self, access_log: FileAccessLog | None = None) -> None:
        self._log = access_log or default_log

    def run(
        self,
        path: str,
        offset: int = 1,
        limit: int | None = None,
        force: bool = False,
    ) -> ToolResult:
        resolved = _resolve_path(path)
        if isinstance(resolved, ToolResult):
            return resolved
        target = resolved

        try:
            lines = target.read_text().splitlines()
        except OSError as exc:
            return ToolResult(
                output=f"Error reading {path}: {exc}", is_error=True
            )

        start = max(offset - 1, 0)
        end = start + limit if limit else len(lines)
        numbered = [
            f"{i + start + 1}\t{line}"
            for i, line in enumerate(lines[start:end])
        ]
        content = "\n".join(numbered)
        content_hash = hashlib.sha256(content.encode()).hexdigest()

        last = self._log.last_record(path)
        unchanged = (
            not force
            and last is not None
            and last.action == "read"
            and last.offset == offset
            and last.limit == limit
            and last.content_hash == content_hash
        )

        self._log.record(
            path,
            "read",
            content_hash=content_hash,
            size=len(content),
            offset=offset,
            limit=limit,
        )

        if unchanged:
            span = f" (lines {offset}-{end})" if limit else ""
            return ToolResult(
                output=f"(unchanged since you last read {path}{span} — pass "
                "force=true to see it again)"
            )

        return ToolResult(output=truncate(content))

description = "Read a text file's contents, optionally a line range. If you already read the exact same path and range in this session and it hasn't changed on disk since, this returns a short notice instead of repeating the content, to save tokens — pass force=true to get the full content again (e.g. if it scrolled out of your context after compaction)." class-attribute instance-attribute

name = 'read_file' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'path': {'type': 'string', 'description': 'File path to read'}, 'offset': {'type': 'integer', 'description': '1-indexed start line'}, 'limit': {'type': 'integer', 'description': 'Max lines to read'}, 'force': {'type': 'boolean', 'description': 'Show full content even if unchanged since your last read.', 'default': False}}, 'required': ['path']} class-attribute instance-attribute

__init__(access_log=None)

Source code in pycodeloop/tools/filesystem.py 
71
72
def __init__(self, access_log: FileAccessLog | None = None) -> None:
    self._log = access_log or default_log

run(path, offset=1, limit=None, force=False)

Source code in pycodeloop/tools/filesystem.py 
 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
def run(
    self,
    path: str,
    offset: int = 1,
    limit: int | None = None,
    force: bool = False,
) -> ToolResult:
    resolved = _resolve_path(path)
    if isinstance(resolved, ToolResult):
        return resolved
    target = resolved

    try:
        lines = target.read_text().splitlines()
    except OSError as exc:
        return ToolResult(
            output=f"Error reading {path}: {exc}", is_error=True
        )

    start = max(offset - 1, 0)
    end = start + limit if limit else len(lines)
    numbered = [
        f"{i + start + 1}\t{line}"
        for i, line in enumerate(lines[start:end])
    ]
    content = "\n".join(numbered)
    content_hash = hashlib.sha256(content.encode()).hexdigest()

    last = self._log.last_record(path)
    unchanged = (
        not force
        and last is not None
        and last.action == "read"
        and last.offset == offset
        and last.limit == limit
        and last.content_hash == content_hash
    )

    self._log.record(
        path,
        "read",
        content_hash=content_hash,
        size=len(content),
        offset=offset,
        limit=limit,
    )

    if unchanged:
        span = f" (lines {offset}-{end})" if limit else ""
        return ToolResult(
            output=f"(unchanged since you last read {path}{span} — pass "
            "force=true to see it again)"
        )

    return ToolResult(output=truncate(content))

pycodeloop.tools.filesystem.WriteFileTool

Bases: Tool

Source code in pycodeloop/tools/filesystem.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
class WriteFileTool(Tool):
    name = "write_file"
    description = "Write content to a file, creating or overwriting it."
    parameters = {
        "type": "object",
        "properties": {
            "path": {"type": "string"},
            "content": {"type": "string"},
        },
        "required": ["path", "content"],
    }
    dangerous = True

    def __init__(self, access_log: FileAccessLog | None = None) -> None:
        self._log = access_log or default_log

    def preview(self, path: str, content: str, **_) -> str:
        if _looks_like_diff(content):
            return (
                "content looks like a unified diff (has a '@@ ... @@' hunk "
                "header or '---'/'+++' file headers), not the file's actual "
                "text. Pass the full literal file content instead."
            )

        try:
            target = resolve_in_workspace(path)
        except OutsideWorkspaceError as exc:
            return str(exc)

        try:
            before = target.read_text()
        except OSError:
            before = ""
        return _diff(path, before, content)

    def run(self, path: str, content: str) -> ToolResult:
        if _looks_like_diff(content):
            return ToolResult(
                output=(
                    "content looks like a unified diff (has a '@@ ... @@' hunk "
                    "header or '---'/'+++' file headers), not the file's actual "
                    "text. Pass the full literal file content instead."
                ),
                is_error=True,
            )

        resolved = _resolve_path(path)
        if isinstance(resolved, ToolResult):
            return resolved
        target = resolved

        try:
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_text(content)
        except OSError as exc:
            return ToolResult(
                output=f"Error writing {path}: {exc}", is_error=True
            )

        self._log.record(
            path,
            "write",
            content_hash=hashlib.sha256(content.encode()).hexdigest(),
            size=len(content),
        )

        return ToolResult(output=f"Wrote {len(content)} bytes to {path}")

dangerous = True class-attribute instance-attribute

description = 'Write content to a file, creating or overwriting it.' class-attribute instance-attribute

name = 'write_file' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'path': {'type': 'string'}, 'content': {'type': 'string'}}, 'required': ['path', 'content']} class-attribute instance-attribute

__init__(access_log=None)

Source code in pycodeloop/tools/filesystem.py 
144
145
def __init__(self, access_log: FileAccessLog | None = None) -> None:
    self._log = access_log or default_log

preview(path, content, **_)

Source code in pycodeloop/tools/filesystem.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def preview(self, path: str, content: str, **_) -> str:
    if _looks_like_diff(content):
        return (
            "content looks like a unified diff (has a '@@ ... @@' hunk "
            "header or '---'/'+++' file headers), not the file's actual "
            "text. Pass the full literal file content instead."
        )

    try:
        target = resolve_in_workspace(path)
    except OutsideWorkspaceError as exc:
        return str(exc)

    try:
        before = target.read_text()
    except OSError:
        before = ""
    return _diff(path, before, content)

run(path, content)

Source code in pycodeloop/tools/filesystem.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def run(self, path: str, content: str) -> ToolResult:
    if _looks_like_diff(content):
        return ToolResult(
            output=(
                "content looks like a unified diff (has a '@@ ... @@' hunk "
                "header or '---'/'+++' file headers), not the file's actual "
                "text. Pass the full literal file content instead."
            ),
            is_error=True,
        )

    resolved = _resolve_path(path)
    if isinstance(resolved, ToolResult):
        return resolved
    target = resolved

    try:
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(content)
    except OSError as exc:
        return ToolResult(
            output=f"Error writing {path}: {exc}", is_error=True
        )

    self._log.record(
        path,
        "write",
        content_hash=hashlib.sha256(content.encode()).hexdigest(),
        size=len(content),
    )

    return ToolResult(output=f"Wrote {len(content)} bytes to {path}")

pycodeloop.tools.filesystem.EditFileTool

Bases: Tool

Source code in pycodeloop/tools/filesystem.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
class EditFileTool(Tool):
    name = "edit_file"
    description = "Replace an exact substring in a file with a new one."
    parameters = {
        "type": "object",
        "properties": {
            "path": {"type": "string"},
            "old_string": {"type": "string"},
            "new_string": {"type": "string"},
            "replace_all": {"type": "boolean", "default": False},
        },
        "required": ["path", "old_string", "new_string"],
    }
    dangerous = True

    def __init__(self, access_log: FileAccessLog | None = None) -> None:
        self._log = access_log or default_log

    def _apply(
        self, path: str, old_string: str, new_string: str, replace_all: bool
    ) -> tuple[Path, str, str] | ToolResult:
        if _looks_like_diff(new_string):
            return ToolResult(
                output=(
                    "new_string looks like a unified diff (has a '@@ ... @@' "
                    "hunk header or '---'/'+++' file headers), not literal "
                    "replacement text. Pass the actual code to substitute in, "
                    "not a diff of it."
                ),
                is_error=True,
            )

        resolved = _resolve_path(path)
        if isinstance(resolved, ToolResult):
            return resolved
        target = resolved

        try:
            text = target.read_text()
        except OSError as exc:
            return ToolResult(
                output=f"Error reading {path}: {exc}", is_error=True
            )

        count = text.count(old_string)

        if count == 0:
            return ToolResult(
                output=f"old_string not found in {path}", is_error=True
            )

        if count > 1 and not replace_all:
            return ToolResult(
                output=(
                    f"old_string is not unique in {path} "
                    f"({count} matches); pass replace_all=true or give "
                    "more context"
                ),
                is_error=True,
            )

        new_text = (
            text.replace(old_string, new_string)
            if replace_all
            else text.replace(old_string, new_string, 1)
        )

        return target, text, new_text

    def preview(
        self,
        path: str,
        old_string: str,
        new_string: str,
        replace_all: bool = False,
        **_,
    ) -> str:
        result = self._apply(path, old_string, new_string, replace_all)

        if isinstance(result, ToolResult):
            return result.output

        _target, before, after = result

        return _diff(path, before, after)

    def run(
        self,
        path: str,
        old_string: str,
        new_string: str,
        replace_all: bool = False,
    ) -> ToolResult:
        result = self._apply(path, old_string, new_string, replace_all)

        if isinstance(result, ToolResult):
            return result

        target, _before, after = result
        target.write_text(after)

        self._log.record(
            path,
            "edit",
            content_hash=hashlib.sha256(after.encode()).hexdigest(),
            size=len(after),
        )

        return ToolResult(output=f"Edited {path}")

dangerous = True class-attribute instance-attribute

description = 'Replace an exact substring in a file with a new one.' class-attribute instance-attribute

name = 'edit_file' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'path': {'type': 'string'}, 'old_string': {'type': 'string'}, 'new_string': {'type': 'string'}, 'replace_all': {'type': 'boolean', 'default': False}}, 'required': ['path', 'old_string', 'new_string']} class-attribute instance-attribute

__init__(access_log=None)

Source code in pycodeloop/tools/filesystem.py 
215
216
def __init__(self, access_log: FileAccessLog | None = None) -> None:
    self._log = access_log or default_log

preview(path, old_string, new_string, replace_all=False, **_)

Source code in pycodeloop/tools/filesystem.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def preview(
    self,
    path: str,
    old_string: str,
    new_string: str,
    replace_all: bool = False,
    **_,
) -> str:
    result = self._apply(path, old_string, new_string, replace_all)

    if isinstance(result, ToolResult):
        return result.output

    _target, before, after = result

    return _diff(path, before, after)

run(path, old_string, new_string, replace_all=False)

Source code in pycodeloop/tools/filesystem.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def run(
    self,
    path: str,
    old_string: str,
    new_string: str,
    replace_all: bool = False,
) -> ToolResult:
    result = self._apply(path, old_string, new_string, replace_all)

    if isinstance(result, ToolResult):
        return result

    target, _before, after = result
    target.write_text(after)

    self._log.record(
        path,
        "edit",
        content_hash=hashlib.sha256(after.encode()).hexdigest(),
        size=len(after),
    )

    return ToolResult(output=f"Edited {path}")

pycodeloop.tools.filesystem.DeleteFileTool

Bases: Tool

Source code in pycodeloop/tools/filesystem.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
class DeleteFileTool(Tool):
    name = "delete_file"
    description = "Delete a file."
    parameters = {
        "type": "object",
        "properties": {"path": {"type": "string"}},
        "required": ["path"],
    }
    dangerous = True

    def __init__(self, access_log: FileAccessLog | None = None) -> None:
        self._log = access_log or default_log

    def preview(self, path: str, **_) -> str:
        try:
            target = resolve_in_workspace(path)
        except OutsideWorkspaceError as exc:
            return str(exc)

        try:
            before = target.read_text()
        except OSError as exc:
            return f"Error reading {path}: {exc}"
        return _diff(path, before, "")

    def run(self, path: str) -> ToolResult:
        resolved = _resolve_path(path)
        if isinstance(resolved, ToolResult):
            return resolved
        target = resolved

        try:
            target.unlink()
        except OSError as exc:
            return ToolResult(
                output=f"Error deleting {path}: {exc}", is_error=True
            )

        self._log.record(path, "delete")

        return ToolResult(output=f"Deleted {path}")

dangerous = True class-attribute instance-attribute

description = 'Delete a file.' class-attribute instance-attribute

name = 'delete_file' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'path': {'type': 'string'}}, 'required': ['path']} class-attribute instance-attribute

__init__(access_log=None)

Source code in pycodeloop/tools/filesystem.py 
321
322
def __init__(self, access_log: FileAccessLog | None = None) -> None:
    self._log = access_log or default_log

preview(path, **_)

Source code in pycodeloop/tools/filesystem.py 
324
325
326
327
328
329
330
331
332
333
334
def preview(self, path: str, **_) -> str:
    try:
        target = resolve_in_workspace(path)
    except OutsideWorkspaceError as exc:
        return str(exc)

    try:
        before = target.read_text()
    except OSError as exc:
        return f"Error reading {path}: {exc}"
    return _diff(path, before, "")

run(path)

Source code in pycodeloop/tools/filesystem.py 
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
def run(self, path: str) -> ToolResult:
    resolved = _resolve_path(path)
    if isinstance(resolved, ToolResult):
        return resolved
    target = resolved

    try:
        target.unlink()
    except OSError as exc:
        return ToolResult(
            output=f"Error deleting {path}: {exc}", is_error=True
        )

    self._log.record(path, "delete")

    return ToolResult(output=f"Deleted {path}")

pycodeloop.tools.filesystem.ListDirTool

Bases: Tool

Source code in pycodeloop/tools/filesystem.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
class ListDirTool(Tool):
    name = "list_dir"
    description = "List files and directories at a given path."
    parameters = {
        "type": "object",
        "properties": {"path": {"type": "string", "default": "."}},
    }

    def run(self, path: str = ".") -> ToolResult:
        resolved = _resolve_path(path)
        if isinstance(resolved, ToolResult):
            return resolved
        target = resolved

        try:
            entries = sorted(target.iterdir())
        except OSError as exc:
            return ToolResult(
                output=f"Error listing {path}: {exc}", is_error=True
            )

        lines = [f"{'d' if e.is_dir() else 'f'} {e.name}" for e in entries]

        return ToolResult(output="\n".join(lines))

description = 'List files and directories at a given path.' class-attribute instance-attribute

name = 'list_dir' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'path': {'type': 'string', 'default': '.'}}} class-attribute instance-attribute

run(path='.')

Source code in pycodeloop/tools/filesystem.py 
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def run(self, path: str = ".") -> ToolResult:
    resolved = _resolve_path(path)
    if isinstance(resolved, ToolResult):
        return resolved
    target = resolved

    try:
        entries = sorted(target.iterdir())
    except OSError as exc:
        return ToolResult(
            output=f"Error listing {path}: {exc}", is_error=True
        )

    lines = [f"{'d' if e.is_dir() else 'f'} {e.name}" for e in entries]

    return ToolResult(output="\n".join(lines))

pycodeloop.tools.search.GlobTool

Bases: Tool

Source code in pycodeloop/tools/search.py 
 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
class GlobTool(Tool):
    name = "glob"
    description = "Find files by glob pattern (e.g. '**/*.py')."
    parameters = {
        "type": "object",
        "properties": {
            "pattern": {"type": "string"},
            "path": {"type": "string", "default": "."},
            "max_results": {"type": "integer", "default": 100},
        },
        "required": ["pattern"],
    }

    def run(
        self, pattern: str, path: str = ".", max_results: int = 100
    ) -> ToolResult:
        try:
            root = resolve_in_workspace(path)
        except OutsideWorkspaceError as exc:
            return ToolResult(output=str(exc), is_error=True)

        try:
            matches = [
                str(p)
                for p in root.glob(pattern)
                if not set(p.parts) & _SKIP_DIRS
            ]
        except (OSError, ValueError) as exc:
            return ToolResult(output=f"Invalid glob: {exc}", is_error=True)

        matches = sorted(matches)[:max_results]

        return ToolResult(
            output="\n".join(matches) if matches else "No matches."
        )

description = "Find files by glob pattern (e.g. '**/*.py')." class-attribute instance-attribute

name = 'glob' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'pattern': {'type': 'string'}, 'path': {'type': 'string', 'default': '.'}, 'max_results': {'type': 'integer', 'default': 100}}, 'required': ['pattern']} class-attribute instance-attribute

run(pattern, path='.', max_results=100)

Source code in pycodeloop/tools/search.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def run(
    self, pattern: str, path: str = ".", max_results: int = 100
) -> ToolResult:
    try:
        root = resolve_in_workspace(path)
    except OutsideWorkspaceError as exc:
        return ToolResult(output=str(exc), is_error=True)

    try:
        matches = [
            str(p)
            for p in root.glob(pattern)
            if not set(p.parts) & _SKIP_DIRS
        ]
    except (OSError, ValueError) as exc:
        return ToolResult(output=f"Invalid glob: {exc}", is_error=True)

    matches = sorted(matches)[:max_results]

    return ToolResult(
        output="\n".join(matches) if matches else "No matches."
    )

pycodeloop.tools.search.GrepTool

Bases: Tool

Source code in pycodeloop/tools/search.py 
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
class GrepTool(Tool):
    name = "grep"
    description = "Search for a regex pattern across files under a path."
    parameters = {
        "type": "object",
        "properties": {
            "pattern": {"type": "string"},
            "path": {"type": "string", "default": "."},
            "max_results": {"type": "integer", "default": 100},
        },
        "required": ["pattern"],
    }

    def run(
        self, pattern: str, path: str = ".", max_results: int = 100
    ) -> ToolResult:
        try:
            root = resolve_in_workspace(path)
        except OutsideWorkspaceError as exc:
            return ToolResult(output=str(exc), is_error=True)

        try:
            regex = re.compile(pattern)
        except re.error as exc:
            return ToolResult(output=f"Invalid regex: {exc}", is_error=True)

        matches: list[str] = []
        for file_path in root.rglob("*"):
            if not file_path.is_file() or set(file_path.parts) & _SKIP_DIRS:
                continue
            if _is_binary(file_path):
                continue
            try:
                text = file_path.read_text(errors="ignore")
            except OSError:
                continue
            for lineno, line in enumerate(text.splitlines(), start=1):
                if regex.search(line):
                    matches.append(f"{file_path}:{lineno}: {line.strip()}")
                    if len(matches) >= max_results:
                        return ToolResult(output=truncate("\n".join(matches)))

        return ToolResult(
            output=truncate("\n".join(matches)) if matches else "No matches."
        )

description = 'Search for a regex pattern across files under a path.' class-attribute instance-attribute

name = 'grep' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'pattern': {'type': 'string'}, 'path': {'type': 'string', 'default': '.'}, 'max_results': {'type': 'integer', 'default': 100}}, 'required': ['pattern']} class-attribute instance-attribute

run(pattern, path='.', max_results=100)

Source code in pycodeloop/tools/search.py 
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
def run(
    self, pattern: str, path: str = ".", max_results: int = 100
) -> ToolResult:
    try:
        root = resolve_in_workspace(path)
    except OutsideWorkspaceError as exc:
        return ToolResult(output=str(exc), is_error=True)

    try:
        regex = re.compile(pattern)
    except re.error as exc:
        return ToolResult(output=f"Invalid regex: {exc}", is_error=True)

    matches: list[str] = []
    for file_path in root.rglob("*"):
        if not file_path.is_file() or set(file_path.parts) & _SKIP_DIRS:
            continue
        if _is_binary(file_path):
            continue
        try:
            text = file_path.read_text(errors="ignore")
        except OSError:
            continue
        for lineno, line in enumerate(text.splitlines(), start=1):
            if regex.search(line):
                matches.append(f"{file_path}:{lineno}: {line.strip()}")
                if len(matches) >= max_results:
                    return ToolResult(output=truncate("\n".join(matches)))

    return ToolResult(
        output=truncate("\n".join(matches)) if matches else "No matches."
    )

pycodeloop.tools.bash.BashTool

Bases: Tool

Source code in pycodeloop/tools/bash.py 
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
class BashTool(Tool):
    name = "bash"
    description = (
        "Run a shell command and return its stdout/stderr. Not limited to "
        "the workspace root — unlike read_file/write_file/grep/etc, this "
        "runs an arbitrary command with full filesystem access, so use it "
        "for paths outside the project only when the user actually asked "
        "for that."
    )
    parameters = {
        "type": "object",
        "properties": {
            "command": {"type": "string"},
            "timeout": {"type": "integer", "default": 120},
        },
        "required": ["command"],
    }
    dangerous = True

    def preview(self, command: str, **_) -> str:
        return f"$ {command}"

    def run(self, command: str, timeout: int = 120) -> ToolResult:
        try:
            proc = subprocess.run(
                command,
                shell=True,
                capture_output=True,
                text=True,
                timeout=timeout,
            )
        except subprocess.TimeoutExpired:
            return ToolResult(
                output=f"Command timed out after {timeout}s", is_error=True
            )

        output = truncate(proc.stdout + proc.stderr)

        return ToolResult(output=output, is_error=proc.returncode != 0)

dangerous = True class-attribute instance-attribute

description = 'Run a shell command and return its stdout/stderr. Not limited to the workspace root — unlike read_file/write_file/grep/etc, this runs an arbitrary command with full filesystem access, so use it for paths outside the project only when the user actually asked for that.' class-attribute instance-attribute

name = 'bash' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'command': {'type': 'string'}, 'timeout': {'type': 'integer', 'default': 120}}, 'required': ['command']} class-attribute instance-attribute

preview(command, **_)

Source code in pycodeloop/tools/bash.py 
30
31
def preview(self, command: str, **_) -> str:
    return f"$ {command}"

run(command, timeout=120)

Source code in pycodeloop/tools/bash.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def run(self, command: str, timeout: int = 120) -> ToolResult:
    try:
        proc = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        return ToolResult(
            output=f"Command timed out after {timeout}s", is_error=True
        )

    output = truncate(proc.stdout + proc.stderr)

    return ToolResult(output=output, is_error=proc.returncode != 0)

pycodeloop.tools.web.WebFetchTool

Bases: Tool

Source code in pycodeloop/tools/web.py 
 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
class WebFetchTool(Tool):
    name = "web_fetch"
    description = "Fetch a URL and return its content as plain text."
    parameters = {
        "type": "object",
        "properties": {
            "url": {"type": "string"},
            "timeout": {"type": "number", "default": 30},
        },
        "required": ["url"],
    }

    def run(self, url: str, timeout: float = 30) -> ToolResult:
        parsed = urlparse(url)
        if parsed.scheme.lower() not in {"http", "https"}:
            return ToolResult(
                output=f"Refused to fetch {url}: only http/https URLs are allowed",
                is_error=True,
            )

        hostname = parsed.hostname
        if not hostname or is_blocked_host(hostname):
            return ToolResult(
                output=f"Refused to fetch {url}: host is not a public address",
                is_error=True,
            )

        try:
            response = safe_request(
                "GET", url, timeout=timeout, follow_redirects=False
            )
        except BlockedHostError:
            return ToolResult(
                output=f"Refused to fetch {url}: host is not a public address",
                is_error=True,
            )
        except httpx.HTTPError as exc:
            return ToolResult(
                output=f"Error fetching {url}: {exc}", is_error=True
            )

        if response.is_redirect:
            return ToolResult(
                output=f"{url} redirects to "
                f"{response.headers.get('location')} — fetch that URL "
                "directly if it's safe to follow",
                is_error=True,
            )

        try:
            response.raise_for_status()
        except httpx.HTTPError as exc:
            return ToolResult(
                output=f"Error fetching {url}: {exc}", is_error=True
            )

        content_type = response.headers.get("content-type", "")
        text = (
            _html_to_text(response.text)
            if "html" in content_type
            else response.text
        )

        return ToolResult(output=truncate(text))

description = 'Fetch a URL and return its content as plain text.' class-attribute instance-attribute

name = 'web_fetch' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'url': {'type': 'string'}, 'timeout': {'type': 'number', 'default': 30}}, 'required': ['url']} class-attribute instance-attribute

run(url, timeout=30)

Source code in pycodeloop/tools/web.py 
 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
def run(self, url: str, timeout: float = 30) -> ToolResult:
    parsed = urlparse(url)
    if parsed.scheme.lower() not in {"http", "https"}:
        return ToolResult(
            output=f"Refused to fetch {url}: only http/https URLs are allowed",
            is_error=True,
        )

    hostname = parsed.hostname
    if not hostname or is_blocked_host(hostname):
        return ToolResult(
            output=f"Refused to fetch {url}: host is not a public address",
            is_error=True,
        )

    try:
        response = safe_request(
            "GET", url, timeout=timeout, follow_redirects=False
        )
    except BlockedHostError:
        return ToolResult(
            output=f"Refused to fetch {url}: host is not a public address",
            is_error=True,
        )
    except httpx.HTTPError as exc:
        return ToolResult(
            output=f"Error fetching {url}: {exc}", is_error=True
        )

    if response.is_redirect:
        return ToolResult(
            output=f"{url} redirects to "
            f"{response.headers.get('location')} — fetch that URL "
            "directly if it's safe to follow",
            is_error=True,
        )

    try:
        response.raise_for_status()
    except httpx.HTTPError as exc:
        return ToolResult(
            output=f"Error fetching {url}: {exc}", is_error=True
        )

    content_type = response.headers.get("content-type", "")
    text = (
        _html_to_text(response.text)
        if "html" in content_type
        else response.text
    )

    return ToolResult(output=truncate(text))

pycodeloop.tools.http_request.HttpRequestTool

Bases: Tool

Source code in pycodeloop/tools/http_request.py 
 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
class HttpRequestTool(Tool):
    name = "http_request"
    description = (
        "Call a JSON HTTP API — any method, headers, and body. Use "
        "web_fetch instead for reading a webpage's text."
    )
    parameters = {
        "type": "object",
        "properties": {
            "url": {"type": "string"},
            "method": {"type": "string", "default": "GET"},
            "headers": {"type": "object"},
            "json_body": {
                "type": "object",
                "description": "Sent as the JSON request body.",
            },
            "timeout": {"type": "number", "default": 30},
        },
        "required": ["url"],
    }
    dangerous = True

    def preview(
        self,
        url: str,
        method: str = "GET",
        headers: dict | None = None,
        json_body: dict | None = None,
        **_,
    ) -> str:
        lines = [f"$ {method.upper()} {url}"]

        if headers:
            lines.append(f"headers: {headers}")

        if json_body is not None:
            lines.append(f"body: {json_body}")

        return "\n".join(lines)

    def run(
        self,
        url: str,
        method: str = "GET",
        headers: dict | None = None,
        json_body: dict | None = None,
        timeout: float = 30,
    ) -> ToolResult:
        method = method.upper()

        if method not in _METHODS:
            return ToolResult(
                output=f"Unsupported method: {method}", is_error=True
            )

        parsed = urlparse(url)
        if parsed.scheme.lower() not in {"http", "https"}:
            return ToolResult(
                output=f"Refused to call {url}: only http/https URLs are allowed",
                is_error=True,
            )

        hostname = parsed.hostname
        if not hostname or is_blocked_host(hostname):
            return ToolResult(
                output=f"Refused to call {url}: host is not a public address",
                is_error=True,
            )

        try:
            response = safe_request(
                method,
                url,
                headers=headers,
                json=json_body,
                timeout=timeout,
                follow_redirects=False,
            )
        except BlockedHostError:
            return ToolResult(
                output=f"Refused to call {url}: host is not a public address",
                is_error=True,
            )
        except httpx.HTTPError as exc:
            return ToolResult(
                output=f"Error calling {url}: {exc}", is_error=True
            )

        if response.is_redirect:
            return ToolResult(
                output=f"{url} redirects to "
                f"{response.headers.get('location')} — call that URL "
                "directly if it's safe to follow",
                is_error=True,
            )

        summary = (
            f"{response.status_code} {response.reason_phrase}\n{response.text}"
        )
        summary = truncate(summary)

        return ToolResult(output=summary, is_error=response.is_error)

dangerous = True class-attribute instance-attribute

description = "Call a JSON HTTP API — any method, headers, and body. Use web_fetch instead for reading a webpage's text." class-attribute instance-attribute

name = 'http_request' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'url': {'type': 'string'}, 'method': {'type': 'string', 'default': 'GET'}, 'headers': {'type': 'object'}, 'json_body': {'type': 'object', 'description': 'Sent as the JSON request body.'}, 'timeout': {'type': 'number', 'default': 30}}, 'required': ['url']} class-attribute instance-attribute

preview(url, method='GET', headers=None, json_body=None, **_)

Source code in pycodeloop/tools/http_request.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def preview(
    self,
    url: str,
    method: str = "GET",
    headers: dict | None = None,
    json_body: dict | None = None,
    **_,
) -> str:
    lines = [f"$ {method.upper()} {url}"]

    if headers:
        lines.append(f"headers: {headers}")

    if json_body is not None:
        lines.append(f"body: {json_body}")

    return "\n".join(lines)

run(url, method='GET', headers=None, json_body=None, timeout=30)

Source code in pycodeloop/tools/http_request.py 
 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
def run(
    self,
    url: str,
    method: str = "GET",
    headers: dict | None = None,
    json_body: dict | None = None,
    timeout: float = 30,
) -> ToolResult:
    method = method.upper()

    if method not in _METHODS:
        return ToolResult(
            output=f"Unsupported method: {method}", is_error=True
        )

    parsed = urlparse(url)
    if parsed.scheme.lower() not in {"http", "https"}:
        return ToolResult(
            output=f"Refused to call {url}: only http/https URLs are allowed",
            is_error=True,
        )

    hostname = parsed.hostname
    if not hostname or is_blocked_host(hostname):
        return ToolResult(
            output=f"Refused to call {url}: host is not a public address",
            is_error=True,
        )

    try:
        response = safe_request(
            method,
            url,
            headers=headers,
            json=json_body,
            timeout=timeout,
            follow_redirects=False,
        )
    except BlockedHostError:
        return ToolResult(
            output=f"Refused to call {url}: host is not a public address",
            is_error=True,
        )
    except httpx.HTTPError as exc:
        return ToolResult(
            output=f"Error calling {url}: {exc}", is_error=True
        )

    if response.is_redirect:
        return ToolResult(
            output=f"{url} redirects to "
            f"{response.headers.get('location')} — call that URL "
            "directly if it's safe to follow",
            is_error=True,
        )

    summary = (
        f"{response.status_code} {response.reason_phrase}\n{response.text}"
    )
    summary = truncate(summary)

    return ToolResult(output=summary, is_error=response.is_error)

pycodeloop.tools.git.GitStatusTool

Bases: Tool

Source code in pycodeloop/tools/git.py 
34
35
36
37
38
39
40
class GitStatusTool(Tool):
    name = "git_status"
    description = "Show the working tree status (git status --porcelain)."
    parameters = {"type": "object", "properties": {}}

    def run(self) -> ToolResult:
        return _run_git("status", "--porcelain=v1", "--branch")

description = 'Show the working tree status (git status --porcelain).' class-attribute instance-attribute

name = 'git_status' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {}} class-attribute instance-attribute

run()

Source code in pycodeloop/tools/git.py 
39
40
def run(self) -> ToolResult:
    return _run_git("status", "--porcelain=v1", "--branch")

pycodeloop.tools.git.GitDiffTool

Bases: Tool

Source code in pycodeloop/tools/git.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class GitDiffTool(Tool):
    name = "git_diff"
    description = "Show unstaged (or staged) changes as a unified diff."
    parameters = {
        "type": "object",
        "properties": {
            "path": {"type": "string"},
            "staged": {"type": "boolean", "default": False},
        },
    }

    def run(self, path: str = "", staged: bool = False) -> ToolResult:
        args = ["diff"]

        if staged:
            args.append("--staged")

        if path:
            args.extend(["--", path])

        return _run_git(*args)

description = 'Show unstaged (or staged) changes as a unified diff.' class-attribute instance-attribute

name = 'git_diff' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'path': {'type': 'string'}, 'staged': {'type': 'boolean', 'default': False}}} class-attribute instance-attribute

run(path='', staged=False)

Source code in pycodeloop/tools/git.py 
54
55
56
57
58
59
60
61
62
63
def run(self, path: str = "", staged: bool = False) -> ToolResult:
    args = ["diff"]

    if staged:
        args.append("--staged")

    if path:
        args.extend(["--", path])

    return _run_git(*args)

pycodeloop.tools.git.GitLogTool

Bases: Tool

Source code in pycodeloop/tools/git.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class GitLogTool(Tool):
    name = "git_log"
    description = "Show recent commit history, one line per commit."
    parameters = {
        "type": "object",
        "properties": {
            "max_count": {"type": "integer", "default": 20},
            "path": {"type": "string"},
        },
    }

    def run(self, max_count: int = 20, path: str = "") -> ToolResult:
        args = ["log", f"-n{max_count}", "--oneline"]

        if path:
            args.extend(["--", path])

        return _run_git(*args)

description = 'Show recent commit history, one line per commit.' class-attribute instance-attribute

name = 'git_log' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'max_count': {'type': 'integer', 'default': 20}, 'path': {'type': 'string'}}} class-attribute instance-attribute

run(max_count=20, path='')

Source code in pycodeloop/tools/git.py 
77
78
79
80
81
82
83
def run(self, max_count: int = 20, path: str = "") -> ToolResult:
    args = ["log", f"-n{max_count}", "--oneline"]

    if path:
        args.extend(["--", path])

    return _run_git(*args)

pycodeloop.tools.git.GitCommitTool

Bases: Tool

Source code in pycodeloop/tools/git.py 
 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
class GitCommitTool(Tool):
    name = "git_commit"
    description = (
        "Stage specific files and create a commit. `paths` is required — "
        "list every file to stage by name. Never stages the whole tree "
        "(no git add -A/-u)."
    )
    parameters = {
        "type": "object",
        "properties": {
            "message": {"type": "string"},
            "paths": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Files to stage (required; no catch-all).",
            },
        },
        "required": ["message", "paths"],
    }
    dangerous = True

    def preview(
        self, message: str, paths: list[str] | None = None, **_
    ) -> str:
        paths = paths or []
        stat = (
            _run_git("diff", "--stat", "HEAD", "--", *paths) if paths else None
        )
        target = ", ".join(paths) if paths else "(no paths)"
        body = f"\n\n{stat.output}" if stat else ""
        return f"$ git commit -m {message!r} ({target}){body}"

    def run(self, message: str, paths: list[str] | None = None) -> ToolResult:
        if not paths:
            return ToolResult(
                output=(
                    "paths is required — pass the specific files to stage. "
                    "Refusing git add -A."
                ),
                is_error=True,
            )

        cleaned = [p for p in paths if p and p not in {"-A", "-u", "--all"}]
        if not cleaned or cleaned != list(paths):
            return ToolResult(
                output=(
                    "paths must be explicit file paths — "
                    "refusing catch-all flags like -A/-u."
                ),
                is_error=True,
            )

        add_result = _run_git("add", "--", *cleaned)

        if add_result.is_error:
            return add_result

        return _run_git("commit", "-m", message)

dangerous = True class-attribute instance-attribute

description = 'Stage specific files and create a commit. `paths` is required — list every file to stage by name. Never stages the whole tree (no git add -A/-u).' class-attribute instance-attribute

name = 'git_commit' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'message': {'type': 'string'}, 'paths': {'type': 'array', 'items': {'type': 'string'}, 'description': 'Files to stage (required; no catch-all).'}}, 'required': ['message', 'paths']} class-attribute instance-attribute

preview(message, paths=None, **_)

Source code in pycodeloop/tools/git.py 
107
108
109
110
111
112
113
114
115
116
def preview(
    self, message: str, paths: list[str] | None = None, **_
) -> str:
    paths = paths or []
    stat = (
        _run_git("diff", "--stat", "HEAD", "--", *paths) if paths else None
    )
    target = ", ".join(paths) if paths else "(no paths)"
    body = f"\n\n{stat.output}" if stat else ""
    return f"$ git commit -m {message!r} ({target}){body}"

run(message, paths=None)

Source code in pycodeloop/tools/git.py 
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
def run(self, message: str, paths: list[str] | None = None) -> ToolResult:
    if not paths:
        return ToolResult(
            output=(
                "paths is required — pass the specific files to stage. "
                "Refusing git add -A."
            ),
            is_error=True,
        )

    cleaned = [p for p in paths if p and p not in {"-A", "-u", "--all"}]
    if not cleaned or cleaned != list(paths):
        return ToolResult(
            output=(
                "paths must be explicit file paths — "
                "refusing catch-all flags like -A/-u."
            ),
            is_error=True,
        )

    add_result = _run_git("add", "--", *cleaned)

    if add_result.is_error:
        return add_result

    return _run_git("commit", "-m", message)

pycodeloop.tools.env.EnvTool

Bases: Tool

Source code in pycodeloop/tools/env.py 
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
class EnvTool(Tool):
    name = "env"
    description = (
        "Read environment variables. Pass `name` for one variable, or "
        "omit it to list every variable name (values whose name looks "
        "sensitive — SECRET, KEY, TOKEN, PASS/PASSWORD, AUTH, COOKIE, "
        "SESSION, CREDENTIAL, … — or a scheme://user:pass@ credential in "
        "the value, are masked)."
    )
    parameters = {
        "type": "object",
        "properties": {
            "name": {"type": "string"},
        },
    }

    def run(self, name: str = "") -> ToolResult:
        if name:
            value = os.environ.get(name)

            if value is None:
                return ToolResult(output=f"{name} is not set", is_error=True)

            return ToolResult(output=f"{name}={_mask(name, value)}")

        lines = [
            f"{key}={_mask(key, value)}"
            for key, value in sorted(os.environ.items())
        ]

        return ToolResult(output="\n".join(lines))

description = 'Read environment variables. Pass `name` for one variable, or omit it to list every variable name (values whose name looks sensitive — SECRET, KEY, TOKEN, PASS/PASSWORD, AUTH, COOKIE, SESSION, CREDENTIAL, … — or a scheme://user:pass@ credential in the value, are masked).' class-attribute instance-attribute

name = 'env' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'name': {'type': 'string'}}} class-attribute instance-attribute

run(name='')

Source code in pycodeloop/tools/env.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def run(self, name: str = "") -> ToolResult:
    if name:
        value = os.environ.get(name)

        if value is None:
            return ToolResult(output=f"{name} is not set", is_error=True)

        return ToolResult(output=f"{name}={_mask(name, value)}")

    lines = [
        f"{key}={_mask(key, value)}"
        for key, value in sorted(os.environ.items())
    ]

    return ToolResult(output="\n".join(lines))

pycodeloop.tools.sql.SqlSchemaTool

Bases: Tool

Source code in pycodeloop/tools/sql.py 
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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
class SqlSchemaTool(Tool):
    name = "sql_schema"
    description = (
        "List tables in a database, or the columns of one table. Pass a "
        "SQLAlchemy connection `url` (e.g. 'sqlite:///path/to.db', "
        "'postgresql://user:pass@host/db', 'mysql+pymysql://user:pass@"
        "host/db'). Use before sql_query to see what's there."
    )
    parameters = {
        "type": "object",
        "properties": {
            "url": {"type": "string"},
            "table": {
                "type": "string",
                "description": (
                    "Omit to list every table instead of one table's columns."
                ),
            },
        },
        "required": ["url"],
    }
    dangerous = True

    def preview(self, url: str, table: str = "", **_) -> str:
        target = f"table {table}" if table else "all tables"
        return f"$ sql_schema {url} ({target})"

    def run(self, url: str, table: str = "") -> ToolResult:
        err = _validate_url(url)
        if err:
            return ToolResult(output=err, is_error=True)

        try:
            engine = _create_engine(url, read_only=True)
            inspector = inspect(engine)
        except SQLAlchemyError as exc:
            return ToolResult(output=f"Error connecting: {exc}", is_error=True)

        try:
            if table:
                columns = inspector.get_columns(table)
                if not columns:
                    return ToolResult(
                        output=f"No such table: {table}", is_error=True
                    )
                lines = [
                    f"{c['name']}\t{c['type']}"
                    + ("" if c.get("nullable", True) else "\tNOT NULL")
                    for c in columns
                ]
                return ToolResult(output=truncate("\n".join(lines)))

            tables = inspector.get_table_names()
            return ToolResult(
                output=(
                    truncate("\n".join(sorted(tables)))
                    if tables
                    else "(no tables)"
                )
            )
        except SQLAlchemyError as exc:
            return ToolResult(
                output=f"Error inspecting schema: {exc}", is_error=True
            )
        finally:
            engine.dispose()

dangerous = True class-attribute instance-attribute

description = "List tables in a database, or the columns of one table. Pass a SQLAlchemy connection `url` (e.g. 'sqlite:///path/to.db', 'postgresql://user:pass@host/db', 'mysql+pymysql://user:pass@host/db'). Use before sql_query to see what's there." class-attribute instance-attribute

name = 'sql_schema' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'url': {'type': 'string'}, 'table': {'type': 'string', 'description': "Omit to list every table instead of one table's columns."}}, 'required': ['url']} class-attribute instance-attribute

preview(url, table='', **_)

Source code in pycodeloop/tools/sql.py 
139
140
141
def preview(self, url: str, table: str = "", **_) -> str:
    target = f"table {table}" if table else "all tables"
    return f"$ sql_schema {url} ({target})"

run(url, table='')

Source code in pycodeloop/tools/sql.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def run(self, url: str, table: str = "") -> ToolResult:
    err = _validate_url(url)
    if err:
        return ToolResult(output=err, is_error=True)

    try:
        engine = _create_engine(url, read_only=True)
        inspector = inspect(engine)
    except SQLAlchemyError as exc:
        return ToolResult(output=f"Error connecting: {exc}", is_error=True)

    try:
        if table:
            columns = inspector.get_columns(table)
            if not columns:
                return ToolResult(
                    output=f"No such table: {table}", is_error=True
                )
            lines = [
                f"{c['name']}\t{c['type']}"
                + ("" if c.get("nullable", True) else "\tNOT NULL")
                for c in columns
            ]
            return ToolResult(output=truncate("\n".join(lines)))

        tables = inspector.get_table_names()
        return ToolResult(
            output=(
                truncate("\n".join(sorted(tables)))
                if tables
                else "(no tables)"
            )
        )
    except SQLAlchemyError as exc:
        return ToolResult(
            output=f"Error inspecting schema: {exc}", is_error=True
        )
    finally:
        engine.dispose()

pycodeloop.tools.sql.SqlQueryTool

Bases: Tool

Source code in pycodeloop/tools/sql.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
class SqlQueryTool(Tool):
    name = "sql_query"
    description = (
        "Run a single read-only SQL statement (SELECT, WITH, EXPLAIN, "
        "safe PRAGMA, SHOW, or DESCRIBE — no INSERT/UPDATE/DELETE/DDL, "
        "no multiple statements) against a database and return the results "
        "as a tab-separated table. Pass a SQLAlchemy connection `url`. "
        "Use sql_schema first if you don't know the tables/columns yet."
    )
    parameters = {
        "type": "object",
        "properties": {
            "url": {"type": "string"},
            "query": {"type": "string"},
            "max_rows": {"type": "integer", "default": 200},
        },
        "required": ["url", "query"],
    }
    dangerous = True

    def preview(self, url: str, query: str = "", **_) -> str:
        return f"$ sql_query {url}\n{query}"

    def run(self, url: str, query: str, max_rows: int = 200) -> ToolResult:
        err = _validate_url(url)
        if err:
            return ToolResult(output=err, is_error=True)

        if _has_multiple_statements(query):
            return ToolResult(
                output="Only a single SQL statement is allowed.", is_error=True
            )

        if not _is_read_only(query):
            return ToolResult(
                output=(
                    "Only read-only statements are allowed (SELECT, WITH, "
                    "EXPLAIN, safe PRAGMA, SHOW, DESCRIBE)."
                ),
                is_error=True,
            )

        try:
            engine = _create_engine(url, read_only=True)
        except SQLAlchemyError as exc:
            return ToolResult(output=f"Error connecting: {exc}", is_error=True)

        try:
            with engine.connect() as conn:
                result = conn.execute(text(query))
                if not result.returns_rows:
                    return ToolResult(output="(no rows returned)")
                columns = list(result.keys())
                rows = result.fetchmany(max_rows)
        except SQLAlchemyError as exc:
            return ToolResult(
                output=f"Error running query: {exc}", is_error=True
            )
        finally:
            engine.dispose()

        if not rows:
            return ToolResult(output="(no rows)")

        lines = ["\t".join(columns)]
        lines += [
            "\t".join("" if v is None else str(v) for v in row) for row in rows
        ]
        summary = "\n".join(lines)
        if len(rows) >= max_rows:
            summary += f"\n… (capped at {max_rows} rows)"

        return ToolResult(output=truncate(summary))

dangerous = True class-attribute instance-attribute

description = "Run a single read-only SQL statement (SELECT, WITH, EXPLAIN, safe PRAGMA, SHOW, or DESCRIBE — no INSERT/UPDATE/DELETE/DDL, no multiple statements) against a database and return the results as a tab-separated table. Pass a SQLAlchemy connection `url`. Use sql_schema first if you don't know the tables/columns yet." class-attribute instance-attribute

name = 'sql_query' class-attribute instance-attribute

parameters = {'type': 'object', 'properties': {'url': {'type': 'string'}, 'query': {'type': 'string'}, 'max_rows': {'type': 'integer', 'default': 200}}, 'required': ['url', 'query']} class-attribute instance-attribute

preview(url, query='', **_)

Source code in pycodeloop/tools/sql.py 
204
205
def preview(self, url: str, query: str = "", **_) -> str:
    return f"$ sql_query {url}\n{query}"

run(url, query, max_rows=200)

Source code in pycodeloop/tools/sql.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def run(self, url: str, query: str, max_rows: int = 200) -> ToolResult:
    err = _validate_url(url)
    if err:
        return ToolResult(output=err, is_error=True)

    if _has_multiple_statements(query):
        return ToolResult(
            output="Only a single SQL statement is allowed.", is_error=True
        )

    if not _is_read_only(query):
        return ToolResult(
            output=(
                "Only read-only statements are allowed (SELECT, WITH, "
                "EXPLAIN, safe PRAGMA, SHOW, DESCRIBE)."
            ),
            is_error=True,
        )

    try:
        engine = _create_engine(url, read_only=True)
    except SQLAlchemyError as exc:
        return ToolResult(output=f"Error connecting: {exc}", is_error=True)

    try:
        with engine.connect() as conn:
            result = conn.execute(text(query))
            if not result.returns_rows:
                return ToolResult(output="(no rows returned)")
            columns = list(result.keys())
            rows = result.fetchmany(max_rows)
    except SQLAlchemyError as exc:
        return ToolResult(
            output=f"Error running query: {exc}", is_error=True
        )
    finally:
        engine.dispose()

    if not rows:
        return ToolResult(output="(no rows)")

    lines = ["\t".join(columns)]
    lines += [
        "\t".join("" if v is None else str(v) for v in row) for row in rows
    ]
    summary = "\n".join(lines)
    if len(rows) >= max_rows:
        summary += f"\n… (capped at {max_rows} rows)"

    return ToolResult(output=truncate(summary))