Repository navigation
Expand file tree
/
Copy pathcommands_cli.py
More file actions
225 lines (200 loc) · 8 KB
/
Copy pathcommands_cli.py
File metadata and controls
225 lines (200 loc) · 8 KB
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
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
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
"""CLI helpers for plugin management."""
from __future__ import annotations
import argparse
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from collections.abc import Callable
from deepagents_code.plugins import (
add_marketplace_source,
install_plugin,
list_available_plugins,
remove_marketplace,
set_installed_plugin_enabled,
uninstall_plugin,
)
from deepagents_code.plugins.marketplace import (
MarketplaceError,
redact_marketplace_source,
redact_urls_in_text,
)
from deepagents_code.plugins.store import load_marketplace_records
def setup_plugin_parser(
subparsers: Any, # noqa: ANN401 # argparse subparsers uses dynamic typing
*,
make_help_action: Callable[[Callable[[], None]], type[argparse.Action]],
add_output_args: Callable[[argparse.ArgumentParser], None] | None = None,
) -> argparse.ArgumentParser:
"""Set up the `plugin` CLI parser.
Args:
subparsers: Parent argparse subparsers object.
make_help_action: Factory for parser-specific help actions.
add_output_args: Optional callback that adds output-format flags.
Returns:
Plugin command parser.
"""
def _help() -> None:
from deepagents_code.ui import show_plugins_help
show_plugins_help()
parent = argparse.ArgumentParser(add_help=False)
parent.add_argument("-h", "--help", action=make_help_action(_help))
parser = subparsers.add_parser(
"plugin",
aliases=["plugins"],
help="Manage plugins",
add_help=False,
parents=[parent],
)
if add_output_args is not None:
add_output_args(parser)
plugin_sub = parser.add_subparsers(dest="plugin_command")
list_parser = plugin_sub.add_parser("list", aliases=["ls"], help="List plugins")
if add_output_args is not None:
add_output_args(list_parser)
install_parser = plugin_sub.add_parser("install", help="Install a plugin")
install_parser.add_argument("plugin_id")
uninstall_parser = plugin_sub.add_parser("uninstall", help="Uninstall a plugin")
uninstall_parser.add_argument("plugin_id")
enable_parser = plugin_sub.add_parser("enable", help="Enable a plugin")
enable_parser.add_argument("plugin_id")
disable_parser = plugin_sub.add_parser("disable", help="Disable a plugin")
disable_parser.add_argument("plugin_id")
marketplace_parser = plugin_sub.add_parser(
"marketplace", help="Manage plugin marketplaces"
)
marketplace_sub = marketplace_parser.add_subparsers(dest="marketplace_command")
marketplace_list = marketplace_sub.add_parser(
"list", aliases=["ls"], help="List marketplaces"
)
if add_output_args is not None:
add_output_args(marketplace_list)
marketplace_add = marketplace_sub.add_parser("add", help="Add a marketplace")
marketplace_add.add_argument("source")
marketplace_remove = marketplace_sub.add_parser(
"remove", help="Remove a marketplace and uninstall its plugins"
)
marketplace_remove.add_argument("name")
return parser
def _plugin_list_rows() -> list[dict[str, object]]:
return [
{"id": plugin_id, "description": description, "enabled": enabled}
for plugin_id, description, enabled in list_available_plugins()
]
def execute_plugin_command(args: argparse.Namespace) -> str | None:
"""Execute a plugin management command.
Args:
args: Parsed argparse namespace.
Returns:
Text output for slash-command callers, or `None` when output was written.
Raises:
SystemExit: With status 1 when a mutating command fails.
"""
output_format = getattr(args, "output_format", "text")
command = getattr(args, "plugin_command", None)
if command is None:
from deepagents_code.ui import show_plugins_help
show_plugins_help()
return None
if command in {"list", "ls"}:
rows = _plugin_list_rows()
if output_format == "json":
from deepagents_code.output import write_json
write_json("plugin list", rows)
return None
if not rows:
text = "No plugin marketplaces configured."
else:
lines = []
for row in rows:
status = "enabled" if row["enabled"] else "disabled"
lines.append(f"{status} {row['id']} {row['description']}".rstrip())
text = "\n".join(lines)
print(text) # noqa: T201
return text
if command == "install":
try:
instance = install_plugin(args.plugin_id)
except (MarketplaceError, FileNotFoundError, OSError, ValueError) as exc:
text = f"Failed to install {args.plugin_id}: {exc}"
print(text) # noqa: T201
raise SystemExit(1) from exc
details = ""
if instance.version is not None:
details = f" (version: {instance.version})"
text = (
f"Installed plugin {instance.plugin_id}{details}. Run /reload to activate."
)
print(text) # noqa: T201
return text
if command == "uninstall":
uninstall_plugin(args.plugin_id)
text = f"Uninstalled plugin {args.plugin_id}."
print(text) # noqa: T201
return text
if command in {"enable", "disable"}:
enabled = command == "enable"
try:
set_installed_plugin_enabled(args.plugin_id, enabled=enabled)
except (MarketplaceError, OSError, ValueError) as exc:
text = f"Failed to {command} {args.plugin_id}: {exc}"
print(text) # noqa: T201
raise SystemExit(1) from exc
text = f"{command.title()}d plugin {args.plugin_id}."
print(text) # noqa: T201
return text
if command == "marketplace":
marketplace_command = getattr(args, "marketplace_command", None)
if marketplace_command in {"list", "ls"}:
records = load_marketplace_records()
rows = [
{
"name": record.name,
"source_type": record.source_type,
"source": redact_marketplace_source(record.source),
"install_location": (
record.install_location
if record.source_type in {"directory", "file"}
else "<managed cache>"
),
}
for record in records.values()
]
if output_format == "json":
from deepagents_code.output import write_json
write_json("plugin marketplace list", rows)
return None
text = (
"No plugin marketplaces configured."
if not rows
else "\n".join(f"{row['name']} {row['source']}" for row in rows)
)
print(text) # noqa: T201
return text
if marketplace_command == "add":
try:
marketplace = add_marketplace_source(args.source)
except (MarketplaceError, FileNotFoundError, OSError, ValueError) as exc:
source = redact_marketplace_source(args.source)
text = (
f"Failed to add marketplace {source}: "
f"{redact_urls_in_text(str(exc))}"
)
print(text) # noqa: T201
raise SystemExit(1) from exc
text = (
f"Added marketplace {marketplace.name} "
f"({len(marketplace.plugins)} plugin(s))."
)
print(text) # noqa: T201
return text
if marketplace_command == "remove":
removed = remove_marketplace(args.name)
text = (
f"Removed marketplace {args.name} and its installed plugins."
if removed
else f"Marketplace {args.name} is not configured."
)
print(text) # noqa: T201
return text
text = "Usage: plugin {list,install,uninstall,enable,disable,marketplace}"
print(text) # noqa: T201
return text