PlanOpticon

planopticon / video_processor / agent / skills / cli_adapter.py

Source Blame History 99 lines

0981a08…	noreply	1	"""Skill: Push artifacts to external tools via their CLIs."""
0981a08…	noreply	2
0981a08…	noreply	3	import json
0981a08…	noreply	4	import shutil
0981a08…	noreply	5	import subprocess
0981a08…	noreply	6	from typing import List
0981a08…	noreply	7
0981a08…	noreply	8	from video_processor.agent.skills.base import AgentContext, Artifact, Skill, register_skill
0981a08…	noreply	9
0981a08…	noreply	10
0981a08…	noreply	11	def _format_github(artifact: Artifact) -> List[str]:
0981a08…	noreply	12	"""Convert artifact to gh CLI commands."""
0981a08…	noreply	13	items = json.loads(artifact.content) if artifact.format == "json" else []
0981a08…	noreply	14	cmds = []
0981a08…	noreply	15	for item in items:
0981a08…	noreply	16	cmd = f"gh issue create --title {json.dumps(item.get('title', ''))}"
0981a08…	noreply	17	if item.get("body"):
0981a08…	noreply	18	cmd += f" --body {json.dumps(item['body'])}"
0981a08…	noreply	19	for label in item.get("labels", []):
0981a08…	noreply	20	cmd += f" --label {json.dumps(label)}"
0981a08…	noreply	21	cmds.append(cmd)
0981a08…	noreply	22	return cmds
0981a08…	noreply	23
0981a08…	noreply	24
0981a08…	noreply	25	def _format_jira(artifact: Artifact) -> List[str]:
0981a08…	noreply	26	"""Convert artifact to jira-cli commands."""
0981a08…	noreply	27	items = json.loads(artifact.content) if artifact.format == "json" else []
0981a08…	noreply	28	return [
0981a08…	noreply	29	f"jira issue create --summary {json.dumps(item.get('title', ''))}"
0981a08…	noreply	30	f" --description {json.dumps(item.get('body', item.get('description', '')))}"
0981a08…	noreply	31	for item in items
0981a08…	noreply	32	]
0981a08…	noreply	33
0981a08…	noreply	34
0981a08…	noreply	35	def _format_linear(artifact: Artifact) -> List[str]:
0981a08…	noreply	36	"""Convert artifact to linear CLI commands."""
0981a08…	noreply	37	items = json.loads(artifact.content) if artifact.format == "json" else []
0981a08…	noreply	38	return [
0981a08…	noreply	39	f"linear issue create --title {json.dumps(item.get('title', ''))}"
0981a08…	noreply	40	f" --description {json.dumps(item.get('body', item.get('description', '')))}"
0981a08…	noreply	41	for item in items
0981a08…	noreply	42	]
0981a08…	noreply	43
0981a08…	noreply	44
0981a08…	noreply	45	_adapters = {"github": _format_github, "jira": _format_jira, "linear": _format_linear}
0981a08…	noreply	46
0981a08…	noreply	47
0981a08…	noreply	48	def run_commands(commands: List[str], dry_run: bool = True) -> List[dict]:
0981a08…	noreply	49	"""Execute CLI commands. In dry_run mode, just return what would run."""
0981a08…	noreply	50	results = []
0981a08…	noreply	51	for cmd in commands:
0981a08…	noreply	52	if dry_run:
0981a08…	noreply	53	results.append({"command": cmd, "status": "dry_run"})
0981a08…	noreply	54	else:
0981a08…	noreply	55	proc = subprocess.run(cmd, shell=True, capture_output=True, text=True)
0981a08…	noreply	56	results.append(
0981a08…	noreply	57	{
0981a08…	noreply	58	"command": cmd,
0981a08…	noreply	59	"returncode": proc.returncode,
0981a08…	noreply	60	"stdout": proc.stdout.strip(),
0981a08…	noreply	61	"stderr": proc.stderr.strip(),
0981a08…	noreply	62	}
0981a08…	noreply	63	)
0981a08…	noreply	64	return results
0981a08…	noreply	65
0981a08…	noreply	66
0981a08…	noreply	67	class CLIAdapterSkill(Skill):
0981a08…	noreply	68	name = "cli_adapter"
0981a08…	noreply	69	description = "Push artifacts to external tools via their CLIs"
0981a08…	noreply	70
0981a08…	noreply	71	def execute(self, context: AgentContext, **kwargs) -> Artifact:
0981a08…	noreply	72	tool = kwargs.get("tool", "github")
0981a08…	noreply	73	artifact = kwargs.get("artifact")
0981a08…	noreply	74	if artifact is None and context.artifacts:
0981a08…	noreply	75	artifact = context.artifacts[-1]
0981a08…	noreply	76	if artifact is None:
0981a08…	noreply	77	return Artifact(
0981a08…	noreply	78	name="CLI Commands", content="[]", artifact_type="cli_commands", format="json"
0981a08…	noreply	79	)
0981a08…	noreply	80
0981a08…	noreply	81	formatter = _adapters.get(tool)
0981a08…	noreply	82	if formatter is None:
0981a08…	noreply	83	return Artifact(
0981a08…	noreply	84	name="CLI Commands",
0981a08…	noreply	85	content=json.dumps({"error": f"Unknown tool: {tool}"}),
0981a08…	noreply	86	artifact_type="cli_commands",
0981a08…	noreply	87	format="json",
0981a08…	noreply	88	)
0981a08…	noreply	89
0981a08…	noreply	90	cli_name = {"github": "gh", "jira": "jira", "linear": "linear"}[tool]
0981a08…	noreply	91	available = shutil.which(cli_name) is not None
0981a08…	noreply	92	commands = formatter(artifact)
0981a08…	noreply	93	content = json.dumps({"tool": tool, "available": available, "commands": commands}, indent=2)
0981a08…	noreply	94	return Artifact(
0981a08…	noreply	95	name="CLI Commands", content=content, artifact_type="cli_commands", format="json"
0981a08…	noreply	96	)
0981a08…	noreply	97
0981a08…	noreply	98
0981a08…	noreply	99	register_skill(CLIAdapterSkill())

PlanOpticon

Keyboard Shortcuts