Something went wrong. Try again.
Where attention appears to go across agent threads: collector, web explorer, and native iOS client
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101import asyncioimport jsonimport osfrom pathlib import Pathimport shleximport shutilfrom typing import Literal
class OrcaError(RuntimeError): pass
class Orca: def __init__(self, binary=None): self.registry_path = Path.home() / '.config/agents-mcp/owned-terminals.json' self.owned = json.loads(self.registry_path.read_text()) if self.registry_path.exists() else {} self.binary = binary or os.getenv('ORCA_BIN', '/Applications/Orca.app/Contents/Resources/bin/orca')
async def run(self, *args): try: process = await asyncio.create_subprocess_exec(self.binary, *args, '--json', stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) except OSError as exc: raise OrcaError('Orca CLI is unavailable') from exc try: stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=30) except asyncio.TimeoutError: process.kill() await process.wait() raise OrcaError('Orca timed out. A mutation may have been accepted; inspect state before retrying.') try: result = json.loads(stdout) except ValueError as exc: raise OrcaError('Orca returned an invalid response; inspect state before retrying mutations.') from exc if process.returncode or result.get('ok') is not True: raise OrcaError(json.dumps(result.get('error', 'Orca command failed'))) return result.get('result', {})
async def terminals(self): result = await self.run('terminal', 'list', '--limit', '1000') for terminal in result.get('terminals', []): terminal['controllable'] = self.owned.get(terminal.get('handle')) == terminal.get('incarnationId') and not terminal.get('orphaned') and bool(terminal.get('connected')) return result
async def checked(self, handle, incarnation=None): if incarnation is not None and self.owned.get(handle) != incarnation: raise OrcaError('Only agent terminals launched by this MCP can be controlled. Existing Orca terminals are read-only.') result = await self.run('terminal', 'show', '--terminal', handle) terminal = result.get('terminal', {}) if terminal.get('handle') != handle: raise OrcaError('Terminal handle did not match') if incarnation is not None: if terminal.get('incarnationId') != incarnation: raise OrcaError('The terminal process changed. List sessions again before controlling it.') if terminal.get('orphaned') or not terminal.get('connected') or not terminal.get('writable'): raise OrcaError('This terminal is orphaned, disconnected, or not writable. Reconnect it in Orca first.') if terminal.get('agentIdentity') not in ('claude', 'codex'): raise OrcaError('Orca has not identified a live agent in this terminal. Refusing to type into an unverified terminal.') return terminal
async def read(self, handle, cursor=None): await self.checked(handle) args = ['terminal', 'read', '--terminal', handle, '--limit', '200'] if cursor is not None: args += ['--cursor', str(cursor)] return await self.run(*args)
async def send(self, handle, incarnation, prompt): if not prompt.strip() or len(prompt) > 20000: raise OrcaError('Prompt must contain 1–20000 characters') if any(ord(c) < 32 and c not in '\n\t' for c in prompt): raise OrcaError('Terminal control characters are not allowed in prompts') await self.checked(handle, incarnation) return await self.run('terminal', 'send', '--terminal', handle, '--text', prompt, '--enter', '--wait-submit', '5')
async def interrupt(self, handle, incarnation): await self.checked(handle, incarnation) return await self.run('terminal', 'send', '--terminal', handle, '--interrupt')
async def start(self, worktree_id, agent: Literal['claude', 'codex']): if agent not in ('claude', 'codex'): raise OrcaError('Unsupported agent') result = await self.run('worktree', 'list', '--limit', '1000') worktree = next((w for w in result.get('worktrees', []) if w.get('id') == worktree_id), None) if not worktree or worktree.get('isArchived'): raise OrcaError('Choose an existing non-archived worktree from list_workspaces') executable = shutil.which(agent) if not executable: raise OrcaError(f'{agent} executable not found') command = 'exec ' + shlex.quote(str(Path(executable).resolve())) result = await self.run('terminal', 'create', '--worktree', 'id:' + worktree_id, '--command', command, '--title', agent) terminal = result.get('terminal', {}) if not terminal.get('handle') or not terminal.get('incarnationId'): raise OrcaError('Orca did not return a stable terminal identity; inspect Orca before retrying') self.owned[terminal['handle']] = terminal['incarnationId'] self.registry_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) fd = os.open(self.registry_path, os.O_CREAT | os.O_TRUNC | os.O_WRONLY, 0o600) with os.fdopen(fd, 'w') as file: json.dump(self.owned, file) return result