def run(self) -> None:
"""Reads the configuration and runs the chosen command until it ends.
The process exits with status 2 when the configuration is invalid, and with the command's own status after the terminal program, `doctor`, `send`, `-p`, `proposals`, `checkpoints`, `skills`, `packages`, `subagents` and `mcp`.
"""
arguments = command_line.CommandLine().parse(self._argument_list)
log_level = logging.INFO
if arguments.command != 'serve':
log_level = logging.WARNING
logging.basicConfig(level=log_level, format=_LOG_FORMAT)
loader = configuration_loader.ConfigurationLoader(
arguments.configuration
)
try:
configuration = loader.load()
except configuration_loader.ConfigurationError as error:
print(error, file=sys.stderr)
sys.exit(2)
secrets = secrets_module.Secrets()
layout = workspace_layout.WorkspaceLayout(
configuration.workspace_directory
)
desk_names = [desk.name for desk in configuration.desks]
for created in layout.create_missing(desk_names):
logging.getLogger(__name__).info(
'Created %s from its template', created
)
if arguments.command == 'doctor':
sys.exit(doctor.Doctor(configuration, secrets).run())
if arguments.command == 'preflight':
sys.exit(
doctor.Doctor(configuration, secrets, going_live=True).run()
)
if arguments.command == 'send':
client = message_client.MessageClient(
f'http://{configuration.gateway.host}:{configuration.gateway.port}',
secrets.gateway_access_token,
)
sys.exit(asyncio.run(client.send(arguments.desk, arguments.text)))
if arguments.command == 'print':
client = message_client.MessageClient(
f'http://{configuration.gateway.host}:{configuration.gateway.port}',
secrets.gateway_access_token,
output_format=arguments.output_format,
)
text = arguments.print_text
if not sys.stdin.isatty():
piped = sys.stdin.read().strip()
if piped:
text = f'{piped}\n\n{text}'
sys.exit(
asyncio.run(
client.send(arguments.chosen_desk or 'general', text)
)
)
if arguments.command == 'terminal':
sys.exit(
self._terminal(
f'http://{configuration.gateway.host}:{configuration.gateway.port}',
secrets.gateway_access_token,
arguments.chosen_desk,
)
)
if arguments.command == 'proposals':
commands = proposal_commands.ProposalCommands(
proposal_store.ProposalStore(layout)
)
sys.exit(
commands.run(
arguments.proposal_action,
vars(arguments).get('proposal_id'),
)
)
if arguments.command == 'checkpoints':
if arguments.desk not in desk_names:
print(f'No desk named {arguments.desk!r}', file=sys.stderr)
sys.exit(1)
commands = checkpoint_commands.CheckpointCommands(
checkpoints_module.Checkpoints(
layout.checkpoint_repository(arguments.desk),
layout.desk_directory(arguments.desk),
)
)
sys.exit(
commands.run(
arguments.checkpoint_action,
vars(arguments).get('checkpoint_id'),
vars(arguments).get('limit', 20),
)
)
if arguments.command == 'skills':
sandbox = bash_sandbox.BashSandbox(
layout.root,
[],
configuration.sandbox.passed_environment,
)
commands = skill_commands.SkillCommands(
layout,
desk_names,
sandbox.passed_environment(),
)
sys.exit(
commands.run(
arguments.skill_action,
vars(arguments).get('source'),
vars(arguments).get('replace', False),
)
)
packages = installed_packages.InstalledPackages(
layout.packages_directory
)
if arguments.command == 'packages':
commands = package_commands.PackageCommands(packages)
argument = vars(arguments).get('source') or vars(arguments).get(
'name'
)
sys.exit(
commands.run(
arguments.package_action,
argument,
vars(arguments).get('replace', False),
)
)
if arguments.command == 'apps':
commands = app_commands.AppCommands(
app_catalog.AppCatalog(
packages,
systemd_user_units.SystemdUserUnits(),
)
)
if arguments.app_action == 'enable':
sys.exit(asyncio.run(commands.enable(arguments.package)))
if arguments.app_action == 'disable':
sys.exit(asyncio.run(commands.disable(arguments.package)))
sys.exit(asyncio.run(commands.list()))
if arguments.command == 'pairing':
commands = pairing_commands.PairingCommands(
pairing_store.PairingStore(layout.telegram_pairing_file),
configuration.telegram.default_desk,
)
if arguments.pairing_action == 'approve':
sys.exit(commands.approve(arguments.code))
if arguments.pairing_action == 'revoke':
sys.exit(commands.revoke(arguments.chat_id))
sys.exit(commands.list())
if arguments.command == 'subagents':
library = subagent_library.SubagentLibrary(
layout.agents_directory,
packages.agent_roots,
)
sys.exit(subagent_commands.SubagentCommands(library).list())
try:
mcp_servers = mcp_server_configuration.McpConfigurationReader(
configuration.mcp_servers_file
).read_with_packages(packages.mcp_files())
except mcp_server_configuration.McpConfigurationError as error:
print(error, file=sys.stderr)
sys.exit(2)
manager = mcp_client_manager.McpClientManager(mcp_servers)
if arguments.command == 'mcp':
sys.exit(asyncio.run(mcp_commands.McpCommands(manager).list()))
self._serve(configuration, secrets, manager, packages)