#!/usr/bin/env python3 """Python 3.10+, standard library only. Saved requests are deliberate, retryable commands.""" import json import os import sys import uuid import urllib.error import urllib.parse import urllib.request class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def load(path): with open(path, encoding='utf-8') as source: return json.load(source) def save(path, value): descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(descriptor, 'w', encoding='utf-8') as target: json.dump(value, target, ensure_ascii=False, indent=2) target.write('\n') def request(origin, path, body=None, token=None): headers = {'accept': 'application/json', 'user-agent': 'Prooftask-Agent-Client/1.8 (Python)'} if token: if not isinstance(token, str) or not token.startswith('ptag_'): raise ValueError('Set a project Agent credential') headers['authorization'] = 'Bearer ' + token data = None if body is not None: data = json.dumps(body, ensure_ascii=False).encode('utf-8') headers['content-type'] = 'application/json' req = urllib.request.Request(origin + path, data=data, headers=headers) try: with urllib.request.build_opener(NoRedirect).open(req, timeout=30) as response: if response.headers.get_content_type() != 'application/json': raise ValueError('Expected API JSON; preserve the saved request and check connectivity') return json.load(response) except urllib.error.HTTPError as error: code = 'API_REQUEST_REJECTED' if error.headers.get_content_type() == 'application/json': try: code = json.load(error).get('error', code) except (ValueError, AttributeError): pass raise ValueError(f'HTTP {error.code}: {code}. Preserve the saved request; inspect the project before replanning.') from None def main(argv): if not argv: raise ValueError('Use connect-start, connect-poll, capabilities, me, inspect, prepare, prepare-issuance, send or export') origin = os.environ.get('PROOFTASK_ORIGIN', 'https://prooftask.com') parsed = urllib.parse.urlsplit(origin) if parsed.scheme != 'https' or not parsed.hostname or parsed.username or parsed.password or parsed.path or parsed.query or parsed.fragment: raise ValueError('Use a canonical HTTPS origin') stored = load(os.environ['PROOFTASK_AGENT_TOKEN_FILE']) if os.environ.get('PROOFTASK_AGENT_TOKEN_FILE') else None if stored and stored.get('origin') != origin: raise ValueError('Credential file belongs to a different origin') token = os.environ.get('PROOFTASK_AGENT_TOKEN') or (stored or {}).get('token') def api(path, body=None): if not token: raise ValueError('Set PROOFTASK_AGENT_TOKEN_FILE or PROOFTASK_AGENT_TOKEN') return request(origin, '/api/agents/v1' + path, body, token) op, *args = argv if op == 'connect-start' and len(args) == 2: result = request(origin, '/api/agents/v1/connections', {'name': args[0]}) save(args[1], {**result, 'origin': origin}) return {'verificationUrl': origin + result['verificationPath'], 'userCode': result['userCode'], 'expiresAt': result['expiresAt'], 'interval': result['interval']} if op == 'connect-poll' and len(args) == 2: connection = load(args[0]) if connection.get('origin') != origin or connection.get('schema') != 'prooftask.agent-connection/v1': raise ValueError('Connection belongs to a different origin or version') result = request(origin, '/api/agents/v1/connections/poll', {'deviceCode': connection['deviceCode']}) if result['state'] == 'CONNECTED': save(args[1], {'schema': 'prooftask.agent-credential/v1', 'origin': origin, **result}) return {'state': 'CONNECTED', 'message': 'Credential saved privately. Set PROOFTASK_AGENT_TOKEN_FILE to the output file.'} return result if op == 'capabilities' and not args: return request(origin, '/api/agents/v1/capabilities') if op == 'draft' and len(args) == 1: body = load(args[0]) if not isinstance(body.get('requestId'), str): raise ValueError('Save a stable requestId in the input file before preparing a draft') return api('/drafts', body) if op == 'proposal-create' and len(args) == 1: return api('/proposals', load(args[0])) if op == 'proposals' and not args: return api('/proposals') if op in {'proposal-inspect', 'proposal-prepare', 'proposal-send'}: if not args or not args[0].startswith('proposal_'): raise ValueError('Use the exact authorized proposal ID') path = '/proposals/' + urllib.parse.quote(args[0], safe='') if op == 'proposal-inspect' and len(args) == 1: return api(path) if len(args) >= 2 and args[1] in {'update', 'publish', 'respond', 'withdraw', 'handoff'}: if op == 'proposal-prepare' and len(args) == 4: view, body = api(path), load(args[2]) body.setdefault('requestId', str(uuid.uuid4())) save(args[3], {'input': body, 'expectedProposalHead': view['proposalHead'], 'automationRevision': view['automation']['revision']}) return {'state': 'PREPARED', 'message': 'Review before sending. Preserve this file for retries.'} if op == 'proposal-send' and len(args) == 3: return api(path + '/' + args[1], load(args[2])) raise ValueError('Invalid proposal command') if op == 'me' and not args: return api('/me') if not args or not args[0].startswith('delivery_'): raise ValueError('Use the exact project ID from your authorized workspace') path = '/projects/' + urllib.parse.quote(args[0], safe='') if op == 'inspect' and len(args) == 1: return api(path) if op == 'prepare-issuance' and len(args) == 2: view = api(path) if view['project']['state'] != 'BUYER_CONFIRMED' or view['recordIssuance']['state'] == 'ISSUED': raise ValueError('There is no accepted record waiting for issuance') save(args[1], {'expectedEventHash': view['project']['headEventHash'], 'automationRevision': view['automation']['revision'], 'input': {'requestId': str(uuid.uuid4()), 'agreementDigest': view['project']['agreement']['digest'], 'confirmationId': view['project']['review']['id'], 'maxCredits': 1}}) return {'state': 'PREPARED', 'message': 'Review before sending issue-record. No purchase or new acceptance.'} commands = {'respond', 'submit', 'review', 'handoff', 'date-revision', 'concern', 'issue-record'} if op == 'prepare' and len(args) == 4 and args[1] in commands: view, body = api(path), load(args[2]) body.setdefault('requestId', str(uuid.uuid4())) save(args[3], {'input': body, 'expectedEventHash': view['project']['headEventHash'], 'automationRevision': view['automation']['revision']}) return {'state': 'PREPARED', 'message': 'Review before sending. Keep this exact file for retries.'} if op == 'send' and len(args) == 3 and args[1] in commands: return api(path + '/' + args[1], load(args[2])) if op == 'export' and len(args) == 2: view, nonce = api(path), str(uuid.uuid4()) result = api(path + '/exports/result-package', {'expectedAgreementDigest': view['project']['agreement']['digest'], 'expectedEventHash': view['project']['headEventHash'], 'nonce': nonce}) if result.get('schema') != 'prooftask.result-review/v1' or result['status']['body']['nonce'] != nonce: raise ValueError('Export response or nonce mismatch') save(args[1], result) return {'state': 'SAVED', 'message': 'Verify the record with independently trusted issuer keys.'} raise ValueError('Invalid arguments; see the public Agent workflow guide') if __name__ == '__main__': try: print(json.dumps(main(sys.argv[1:]), ensure_ascii=False, indent=2)) except (ValueError, OSError, KeyError) as error: print(str(error), file=sys.stderr) sys.exit(1)