"""Generic self-service bootstrap for the existing single-user MCP example. Registration is one application per environment, not one per user or Agent. The separate private journal retains the original request and optional recovery key. It never contains a Client Secret, a wallet key, or a payment authority. Run explicitly; importing this module performs no network or filesystem work. """ from __future__ import annotations import argparse import asyncio from dataclasses import dataclass import json import logging from pathlib import Path import re import secrets from urllib.parse import urlsplit import httpx import uvicorn from app import ClientError, Config, ReferenceClient, SCOPES, Store, digest, https_origin, make_app @dataclass(frozen=True) class RegistrationConfig: issuer: str resource: str client_name: str redirect_uri: str scopes: tuple[str, ...] networks: tuple[str, ...] subject_context_public_key: str | None = None def __post_init__(self): if (https_origin(self.issuer) != self.issuer or self.resource != self.issuer + "/mcp/third-party" or type(self.client_name) is not str or not 1 <= len(self.client_name) <= 100 or any(ord(char) < 32 or ord(char) == 127 for char in self.client_name) or urlsplit(self.redirect_uri).path in {"", "/"} or urlsplit(self.redirect_uri).path.startswith("/api/") or not self.scopes or not set(self.scopes) <= SCOPES or not self.networks or not set(self.networks) <= {"eip155:80002", "eip155:8453"} or self.subject_context_public_key is not None and not re.fullmatch( r"[A-Za-z0-9_-]{43}", self.subject_context_public_key)): raise ClientError("INVALID_PUBLIC_CONFIGURATION") https_origin(self.redirect_uri) def request(self): value = {"client_name": self.client_name, "redirect_uris": [self.redirect_uri], "grant_types": ["authorization_code", "refresh_token"], "response_types": ["code"], "token_endpoint_auth_method": "none", "scope": " ".join(sorted(self.scopes)), "agentonomy_networks": sorted(self.networks)} if self.subject_context_public_key is not None: value["subject_context_public_key"] = self.subject_context_public_key return value class RegistrationClient: """Explicit registration/recovery; never guesses an ID or loops new requests. The journal port must provide atomic read/change operations. The provided Store is private SQLite for one-user testing; multi-user hosts use their own trusted transactional store and keep this journal apart from OAuth token storage. """ key = "registration:v1" def __init__(self, config, journal, *, verify=True, transport=None, idempotent=True): if type(config) is not RegistrationConfig or verify is False or type(idempotent) is not bool: raise ClientError("INVALID_REGISTRATION_CONFIGURATION") self.config, self.journal = config, journal self.verify, self.transport = verify, transport request = config.request() binding = {"issuer": config.issuer, "resource": config.resource, "request": request, "idempotent": idempotent} self.fingerprint = digest(json.dumps(binding, sort_keys=True, separators=(",", ":"))) def bind(row): if row is not None: if row.get("fingerprint") != self.fingerprint: raise ClientError("REGISTRATION_CONFIGURATION_MISMATCH") return row return {**binding, "fingerprint": self.fingerprint, "status": "prepared", "idempotency_key": secrets.token_urlsafe(32) if idempotent else None} journal.change(self.key, bind) def _configuration(self, row): return Config(self.config.issuer, self.config.resource, row["client_id"], self.config.redirect_uri, tuple(self.config.scopes)) def _response(self, response, request): if len(response.content) > 65536: raise ClientError("REGISTRATION_RESPONSE_INVALID") try: value = response.json() if (type(value) is not dict or not re.fullmatch(r"[A-Za-z0-9_.:-]{1,200}", value.get("client_id", "")) or type(value.get("client_id_issued_at")) is not int or value["client_id_issued_at"] < 0 or any(value.get(key) != item for key, item in request.items()) or any(key in value for key in ("client_secret", "registration_access_token"))): raise ValueError() diagnostic = value.get("agentonomy_registration") if diagnostic is not None and (type(diagnostic) is not dict or diagnostic.get("status") not in {"pending", "ready", "disabled"} or diagnostic.get("version") != 1 or type(diagnostic["version"]) is not int or not re.fullmatch(r"[A-Za-z0-9_.:-]{1,200}", diagnostic.get("operation_id", ""))): raise ValueError() return value except (ValueError, TypeError, KeyError, RecursionError): raise ClientError("REGISTRATION_RESPONSE_INVALID") from None async def register(self, *, recover=False): row = self.journal.read(self.key) if row["status"] == "registered": return self._configuration(row) if row["status"] == "disabled": raise ClientError("REGISTRATION_DISABLED") if row["status"] == "rejected": raise ClientError("REGISTRATION_REJECTED") if row["status"] != "prepared": if row["idempotency_key"] is None: raise ClientError("REGISTRATION_RESULT_UNKNOWN") if not recover: raise ClientError("REGISTRATION_RECOVERY_REQUIRED") async with httpx.AsyncClient(verify=self.verify, transport=self.transport, trust_env=False, follow_redirects=False, timeout=30) as http: try: discovery = await http.get(self.config.issuer + "/.well-known/oauth-authorization-server") metadata = discovery.json() if discovery.status_code == 200 and len(discovery.content) <= 65536 else {} except Exception: raise ClientError("REGISTRATION_DISCOVERY_UNAVAILABLE") from None if (type(metadata) is not dict or metadata.get("issuer") != self.config.issuer or metadata.get("registration_endpoint") != self.config.issuer + "/oauth/register"): raise ClientError("REGISTRATION_DISCOVERY_MISMATCH") def claim(current): if current["fingerprint"] != self.fingerprint: raise ClientError("REGISTRATION_CONFIGURATION_MISMATCH") if current["status"] in {"registered", "disabled", "rejected"}: return current if current["status"] != "prepared": if current["idempotency_key"] is None: raise ClientError("REGISTRATION_RESULT_UNKNOWN") if not recover: raise ClientError("REGISTRATION_RECOVERY_REQUIRED") return {**current, "status": "claimed"} row = self.journal.change(self.key, claim) if row["status"] == "registered": return self._configuration(row) if row["status"] in {"disabled", "rejected"}: raise ClientError("REGISTRATION_" + row["status"].upper()) headers = {"Idempotency-Key": row["idempotency_key"]} if row["idempotency_key"] else {} try: response = await http.post(metadata["registration_endpoint"], headers=headers, json=row["request"]) except Exception: self._record(row, status="unknown") raise ClientError("REGISTRATION_RESULT_UNKNOWN") from None if response.status_code not in {200, 201}: # 5xx/429 may follow acceptance. Keep the exact journal; do not # reinterpret a transport error as permission to register anew. status = "rejected" if response.status_code in {400, 401, 403, 409, 410, 422} else "unknown" self._record(row, status=status) raise ClientError("REGISTRATION_REJECTED" if status == "rejected" else "REGISTRATION_RESULT_UNKNOWN") try: value = self._response(response, row["request"]) except ClientError: self._record(row, status="unknown") raise diagnostic = value.get("agentonomy_registration") state = "disabled" if diagnostic and diagnostic["status"] == "disabled" else "registered" saved = self._record(row, status=state, client_id=value["client_id"], registration=diagnostic) if saved["status"] == "disabled": raise ClientError("REGISTRATION_DISABLED") return self._configuration(saved) def _record(self, expected, **changes): def update(row): if (row["fingerprint"] != expected["fingerprint"] or row["idempotency_key"] != expected["idempotency_key"]): raise ClientError("REGISTRATION_CONFIGURATION_MISMATCH") if row["status"] in {"registered", "disabled", "rejected"}: if changes.get("client_id") not in {None, row.get("client_id")}: raise ClientError("REGISTRATION_RESPONSE_INVALID") return row return {**row, **changes} return self.journal.change(self.key, update) def main(): parser = argparse.ArgumentParser(description="Register once, then run the existing single-user reference backend") parser.add_argument("--config", required=True, type=Path, help="public registration JSON only") parser.add_argument("--registration-state-dir", required=True, type=Path, help="separate private registration journal") parser.add_argument("--state-dir", required=True, type=Path, help="separate private OAuth token store") parser.add_argument("--recover-registration", action="store_true", help="retry ONLY the persisted request and key") parser.add_argument("--port", type=int, default=8787) args = parser.parse_args() logging.disable(logging.CRITICAL) try: if args.registration_state_dir.resolve() == args.state_dir.resolve(): raise ClientError("SEPARATE_REGISTRATION_JOURNAL_REQUIRED") registration = RegistrationConfig(**json.loads(args.config.read_text())) if registration.subject_context_public_key is not None: # This CLI is a single-user browser demo. Strong multi-user binding # requires the host's injected subject signer, not key discovery. raise ClientError("USE_TRUSTED_MULTIUSER_HOST") bootstrap = RegistrationClient(registration, Store(args.registration_state_dir)) config = asyncio.run(bootstrap.register(recover=args.recover_registration)) client = ReferenceClient(config, Store(args.state_dir)) uvicorn.run(make_app(client), host="127.0.0.1", port=args.port, access_log=False, log_config=None, log_level="critical", proxy_headers=False) except ClientError as exc: raise SystemExit("SELF_SERVICE_STOPPED: " + str(exc)) from None except Exception: raise SystemExit("SELF_SERVICE_STOPPED: check public config and private state; details not logged") from None if __name__ == "__main__": main()