[アップデート]AWS Agent Registryが一般提供開始されました！

aws organizations enable-aws-service-access \
  --service-principal agent-registry.amazonaws.com \
  --profile org-management

aws iam create-service-linked-role \
  --aws-service-name agent-registry.amazonaws.com \
  --profile org-management

aws agent-registry-control create-registry \
  --name registry-ga-autodetection \
  --discovery-configuration '{"authorizerType":"AWS_IAM"}' \
  --approval-configuration '{"autoApprovalRules":[]}' \
  --auto-detection-configuration '{"scope":"ORGANIZATION","enabled":true}' \
  --region us-east-1 \
  --profile org-management

export REGISTRY_ARN='<作成結果のregistryArn>'

aws agent-registry-control get-registry \
  --registry-id "$REGISTRY_ARN" \
  --query '{registryStatus:status,autoDetection:autoDetection}' \
  --region us-east-1 --profile org-management

aws agent-registry-control list-registry-records \
  --registry-id "$REGISTRY_ARN" \
  --region us-east-1 --profile org-management

aws agent-registry-control get-registry-record \
  --registry-id "$REGISTRY_ARN" \
  --record-id '<一覧に表示されたレコードID>' \
  --query '{recordType:recordType,status:status,autoDetected:createdByAutoDetection,provenance:provenance}' \
  --region us-east-1 --profile org-management

aws agent-registry-control update-registry \
  --registry-id "$REGISTRY_ARN" \
  --auto-detection-configuration '{"optionalValue":{"scope":"ORGANIZATION","enabled":false}}' \
  --region us-east-1 --profile org-management

# 一覧にあるレコードごとに実行
aws agent-registry-control delete-registry-record \
  --registry-id "$REGISTRY_ARN" \
  --record-id '<削除するレコードID>' \
  --region us-east-1 --profile org-management

aws agent-registry-control delete-registry \
  --registry-id "$REGISTRY_ARN" \
  --region us-east-1 --profile org-management

aws organizations disable-aws-service-access \
  --service-principal agent-registry.amazonaws.com \
  --profile org-management

"""Disposable AWS Agent Registry GA lab; run with uv run scripts/registry_lab.py."""
import argparse
import json
import os
from pathlib import Path
import time
import uuid
import urllib.request
from urllib.parse import quote

import boto3
from botocore.exceptions import ClientError
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

ROOT = Path(__file__).resolve().parents[1]
STATE_PATH = ROOT / '.local/registry-state.json'
EVENTS_PATH = ROOT / '.local/registry-events.jsonl'
SESSION = boto3.Session(region_name='us-east-1')
CONTROL = SESSION.client('agent-registry-control')
DISCOVERY = SESSION.client('agent-registry')
STATE = json.loads(STATE_PATH.read_text()) if STATE_PATH.exists() else {}

def save():
    STATE_PATH.parent.mkdir(mode=0o700, exist_ok=True)
    fd = os.open(STATE_PATH, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, 'w') as stream:
        json.dump(STATE, stream, indent=2, default=str)

def event(name, **values):
    row = {'event': name, **values}
    line = json.dumps(row, ensure_ascii=False, default=str)
    print(line, flush=True)
    with EVENTS_PATH.open('a') as stream:
        stream.write(line + '\n')

def wait_record(record_id, desired='DRAFT'):
    for _ in range(150):
        result = CONTROL.get_registry_record(registryId=STATE['registryId'], recordId=record_id)
        if result['status'] == desired:
            return result
        if result['status'].endswith('FAILED'):
            raise RuntimeError(result.get('statusReason', result['status']))
        time.sleep(4)
    raise TimeoutError(f'Record did not reach {desired}')

def create_registry():
    if 'registryId' not in STATE:
        result = CONTROL.create_registry(
            name='registry-ga-blog-' + uuid.uuid4().hex[:8],
            description='Disposable Agent Registry GA verification',
            discoveryConfiguration={'authorizerType': 'AWS_IAM'},
            approvalConfiguration={'autoApprovalRules': []},
            tags={'Project': 'registry-ga-blog'},
        )
        STATE.update(registryId=result['registryArn'], records={})
        save()
        event('registry_created', http_status=result['ResponseMetadata']['HTTPStatusCode'])
    for attempt in range(150):
        result = CONTROL.get_registry(registryId=STATE['registryId'])
        if result['status'] == 'READY':
            event('registry_ready', status=result['status'], autoDetection=result.get('autoDetection'))
            return
        if result['status'].endswith('FAILED'):
            raise RuntimeError(result.get('statusReason', result['status']))
        if attempt % 8 == 0:
            event('registry_wait', status=result['status'])
        time.sleep(4)
    raise TimeoutError('Registry not ready')

def runtime_url():
    arn = os.environ['AGENT_RUNTIME_ARN']
    region = arn.split(':')[3]
    return f'https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{quote(arn, safe="")}/invocations?qualifier=DEFAULT'

def create_records():
    for kind in ('http',):
        if kind in STATE['records']:
            continue
        result = CONTROL.create_registry_record(
            registryId=STATE['registryId'], name='calculation-' + kind + '-agent',
            displayName='Calculation ' + kind.upper() + ' Agent',
            description='Adds two numbers using the add_numbers tool. Hosted on AgentCore Runtime. '
                        'POST JSON with a prompt string; returns a text/event-stream response. '
                        'Authentication: AWS IAM (SigV4). Owner: demo team.',
            recordType='AGENT', recordVersion='1.0.0',
            descriptors={kind: {'source': {'fromUrl': {'url': runtime_url()}}}},
            tags={'Project': 'registry-ga-blog'},
        )
        record_id = result['recordArn'].rsplit('/', 1)[-1]
        STATE['records'][kind] = record_id
        save()
        event('record_created', kind=kind, status=result['status'])
        record = wait_record(record_id)
        event('record_ready', kind=kind, status=record['status'], recordType=record['recordType'],
              descriptors=record['descriptors'], createdByAutoDetection=record.get('createdByAutoDetection'))

def custom():
    definition = {
        'endpoint': runtime_url(),
        'authentication': {'type': 'AWS_IAM', 'signingService': 'bedrock-agentcore'},
        'request': {'method': 'POST', 'contentType': 'application/json',
                    'example': {'prompt': 'Use add_numbers to calculate 2 + 3.'}},
        'response': {'contentType': 'text/event-stream'},
        'owner': 'demo team',
    }
    result = CONTROL.create_registry_record(
        registryId=STATE['registryId'], name='calculation-custom-agent',
        displayName='Calculation Custom Agent',
        description='Adds two numbers. Connection instructions for application developers.',
        recordType='AGENT', recordVersion='1.0.0',
        descriptors={'custom': {'data': json.dumps(definition)}},
    )
    rid = result['recordArn'].rsplit('/', 1)[-1]
    STATE['records']['custom'] = rid
    save()
    wait_record(rid)
    CONTROL.submit_registry_record_for_approval(registryId=STATE['registryId'], recordId=rid)
    wait_record(rid, 'PENDING_APPROVAL')
    CONTROL.update_registry_record_status(registryId=STATE['registryId'], recordId=rid,
        status='APPROVED', statusReason='Reviewed the example connection instructions.')
    wait_record(rid, 'APPROVED')
    event('custom_approved', recordType='AGENT', descriptor='custom')

def snapshot(label):
    registry_id = STATE['registryId']
    managed = CONTROL.list_registry_records(registryId=registry_id)['registryRecords']
    discovered = DISCOVERY.list_discoverable_registry_records(registryId=registry_id)['registryRecords']
    def summary(records):
        return [{key: record.get(key) for key in ('name', 'status', 'recordType', 'description')} for record in records]
    event('snapshot', label=label, managed=summary(managed), discovered=summary(discovered))

def approve():
    record_id = STATE['records']['http']
    current = CONTROL.get_registry_record(registryId=STATE['registryId'], recordId=record_id)
    if current['status'] == 'DRAFT':
        result = CONTROL.submit_registry_record_for_approval(registryId=STATE['registryId'], recordId=record_id)
        event('submitted', status=result['status'])
        wait_record(record_id, 'PENDING_APPROVAL')
    if current['status'] != 'APPROVED':
        result = CONTROL.update_registry_record_status(
            registryId=STATE['registryId'], recordId=record_id,
            status='APPROVED', statusReason='Reviewed the disposable example metadata.',
        )
        event('approved', status=result['status'])
        wait_record(record_id, 'APPROVED')

def revise():
    CONTROL.update_registry_record(
        registryId=STATE['registryId'], recordId=STATE['records']['http'],
        description={'optionalValue': 'Revised draft: add_numbers accepts two integers.'},
    )
    result = wait_record(STATE['records']['http'])
    event('revised', status=result['status'], description=result['description'])

def batch():
    result = DISCOVERY.batch_get_discoverable_registry_record(entries=[{
        'registryId': STATE['registryId'], 'recordIds': list(STATE['records'].values()),
    }])
    event('batch_get', http_status=result['ResponseMetadata']['HTTPStatusCode'],
          records=[{'name': r['name'], 'descriptors': r['descriptors']} for r in result['registryRecords']],
          errors=[{'kind': next(k for k, v in STATE['records'].items() if v == e['recordId'].rsplit('/', 1)[-1]),
                   'errorCode': e['errorCode']} for e in result['errors']])

def mcp():
    registry_id = STATE['registryId'].rsplit('/', 1)[-1]
    url = f'https://agent-registry.us-east-1.api.aws/registry/{registry_id}/mcp'
    headers = {'Content-Type': 'application/json', 'Accept': 'application/json, text/event-stream'}
    def request(method, params, number):
        body = {'jsonrpc': '2.0', 'method': method, 'params': params}
        if number is not None:
            body['id'] = number
        payload = json.dumps(body)
        signed = AWSRequest(method='POST', url=url, data=payload, headers=headers)
        SigV4Auth(SESSION.get_credentials().get_frozen_credentials(), 'agent-registry', 'us-east-1').add_auth(signed)
        req = urllib.request.Request(url, data=payload.encode(), headers=dict(signed.headers.items()), method='POST')
        with urllib.request.urlopen(req, timeout=60) as response:
            if response.headers.get('Mcp-Session-Id'):
                headers['Mcp-Session-Id'] = response.headers['Mcp-Session-Id']
            raw = response.read().decode()
            if not raw:
                return None
            if raw.startswith('event:') or raw.startswith('data:'):
                raw = next(line[5:].strip() for line in raw.splitlines() if line.startswith('data:'))
            result = json.loads(raw)
            if 'error' in result:
                raise RuntimeError(result['error'])
            if number is None:
                return result
            return result['result']
    result = request('initialize', {'protocolVersion': '2025-11-25', 'capabilities': {},
                                   'clientInfo': {'name': 'registry-ga-blog', 'version': '1.0.0'}}, 1)
    headers['MCP-Protocol-Version'] = result['protocolVersion']
    event('mcp_initialized', protocolVersion=result['protocolVersion'])
    request('notifications/initialized', {}, None)
    result = request('tools/list', {}, 2)
    event('mcp_tools', tools=result['tools'])
    result = request('tools/call', {'name': 'list_discoverable_registry_records', 'arguments': {}}, 3)
    if result.get('isError'):
        raise RuntimeError(result)
    event('mcp_list_called', result=result)
    result = request('tools/call', {'name': 'search_discoverable_registry_records',
        'arguments': {'searchQuery': 'add two numbers calculation', 'maxResults': 5}}, 4)
    if result.get('isError'):
        raise RuntimeError(result)
    event('mcp_search_called', result=result)
    result = request('tools/call', {'name': 'batch_get_discoverable_registry_record',
        'arguments': {'recordIds': list(STATE['records'].values())}}, 5)
    if result.get('isError'):
        raise RuntimeError(result)
    event('mcp_batch_called', result=result)

def invoke():
    result = DISCOVERY.batch_get_discoverable_registry_record(entries=[{
        'registryId': STATE['registryId'], 'recordIds': [STATE['records']['http']],
    }])
    url = result['registryRecords'][0]['descriptors']['http']['source']['fromUrl']['url']
    assert url == runtime_url(), 'Unexpected invocation endpoint'
    payload = json.dumps({'prompt': 'Use only add_numbers to calculate 2 + 3. Reply with the result.'})
    headers = {'Content-Type': 'application/json', 'Accept': 'text/event-stream',
               'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id': str(uuid.uuid4())}
    signed = AWSRequest(method='POST', url=url, data=payload, headers=headers)
    SigV4Auth(SESSION.get_credentials().get_frozen_credentials(), 'bedrock-agentcore', SESSION.region_name).add_auth(signed)
    req = urllib.request.Request(url, data=payload.encode(), headers=dict(signed.headers.items()), method='POST')
    with urllib.request.urlopen(req, timeout=180) as response:
        raw = response.read().decode()
        (ROOT / '.local/runtime-response.txt').write_text(raw)
        text = ''.join(json.loads(line[6:]).get('event', {}).get('contentBlockDelta', {})
                       .get('delta', {}).get('text', '')
                       for line in raw.splitlines() if line.startswith('data: '))
        event('runtime_invoked', http_status=response.status,
              content_type=response.headers.get('Content-Type'), response_text=text)

def cleanup():
    if 'registryId' not in STATE:
        return
    for kind, record_id in list(STATE['records'].items()):
        try:
            CONTROL.delete_registry_record(registryId=STATE['registryId'], recordId=record_id)
        except ClientError as exc:
            if exc.response['Error']['Code'] != 'ResourceNotFoundException':
                raise
        event('record_deleted', kind=kind)
        del STATE['records'][kind]
        save()
    CONTROL.delete_registry(registryId=STATE['registryId'])
    for _ in range(60):
        try:
            CONTROL.get_registry(registryId=STATE['registryId'])
        except ClientError as exc:
            if exc.response['Error']['Code'] == 'ResourceNotFoundException':
                event('registry_deleted')
                STATE.clear()
                save()
                return
            raise
        time.sleep(2)
    raise TimeoutError('Registry deletion not complete')

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('action', choices=['create', 'snapshot', 'approve', 'revise', 'batch', 'mcp', 'invoke', 'custom', 'cleanup'])
    parser.add_argument('--label', default='manual')
    args = parser.parse_args()
    if args.action == 'create':
        create_registry()
        create_records()
    elif args.action == 'snapshot':
        snapshot(args.label)
    else:
        globals()[args.action]()

{
  "event": "snapshot",
  "label": "before-approval",
  "managed": [
    {
      "name": "calculation-http-agent",
      "status": "DRAFT",
      "recordType": "AGENT",
      "description": "Adds two numbers using the add_numbers tool. Hosted on AgentCore Runtime. POST JSON with a prompt string; returns a text/event-stream response. Authentication: AWS IAM (SigV4). Owner: demo team."
    }
  ],
  "discovered": []
}

{
  "event": "snapshot",
  "label": "after-approval",
  "managed": [
    {
      "name": "calculation-http-agent",
      "status": "APPROVED",
      "recordType": "AGENT",
      "description": "Adds two numbers using the add_numbers tool. Hosted on AgentCore Runtime. POST JSON with a prompt string; returns a text/event-stream response. Authentication: AWS IAM (SigV4). Owner: demo team."
    }
  ],
  "discovered": [
    {
      "name": "calculation-http-agent",
      "status": "APPROVED",
      "recordType": "AGENT",
      "description": "Adds two numbers using the add_numbers tool. Hosted on AgentCore Runtime. POST JSON with a prompt string; returns a text/event-stream response. Authentication: AWS IAM (SigV4). Owner: demo team."
    }
  ]
}

{
  "event": "snapshot",
  "label": "after-revision",
  "managed": [
    {
      "name": "calculation-http-agent",
      "status": "DRAFT",
      "recordType": "AGENT",
      "description": "Revised draft: add_numbers accepts two integers."
    }
  ],
  "discovered": [
    {
      "name": "calculation-http-agent",
      "status": "APPROVED",
      "recordType": "AGENT",
      "description": "Adds two numbers using the add_numbers tool. Hosted on AgentCore Runtime. POST JSON with a prompt string; returns a text/event-stream response. Authentication: AWS IAM (SigV4). Owner: demo team."
    }
  ]
}

{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "list_discoverable_registry_records",
    "arguments": {}
  }
}

{
  "registryRecords": [
    {
      "recordArn": "arn:aws:agent-registry:us-east-1:<ACCOUNT_ID>:registry/<REGISTRY_ID>/record/<RECORD_ID>",
      "recordId": "<RECORD_ID>",
      "name": "calculation-http-agent",
      "displayName": "Calculation HTTP Agent",
      "description": "Adds two numbers using the add_numbers tool. Hosted on AgentCore Runtime. POST JSON with a prompt string; returns a text/event-stream response. Authentication: AWS IAM (SigV4). Owner: demo team.",
      "recordType": "AGENT",
      "recordVersion": "1.0.0",
      "status": "APPROVED",
      "createdAt": "2026-09-15T04:48:58.608183099Z",
      "updatedAt": "2026-09-15T04:49:56.904363846Z",
      "descriptorTypes": [
        "http"
      ]
    }
  ]
}

{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "search_discoverable_registry_records",
    "arguments": {
      "searchQuery": "add two numbers calculation",
      "maxResults": 5
    }
  }
}

[
  {
    "recordId": "<RECORD_ID>",
    "name": "calculation-http-agent",
    "displayName": "Calculation HTTP Agent",
    "description": "Adds two numbers using the add_numbers tool. Hosted on AgentCore Runtime. POST JSON with a prompt string; returns a text/event-stream response. Authentication: AWS IAM (SigV4). Owner: demo team.",
    "recordType": "AGENT",
    "descriptors": {
      "mcpServer": null,
      "a2aAgentCard": null,
      "agentSkillsDefinition": null,
      "custom": null,
      "http": {
        "source": {
          "fromUrl": {
            "url": "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-east-1%3A<ACCOUNT_ID>%3Aruntime%2F<RUNTIME_ID>/invocations?qualifier=DEFAULT"
          }
        }
      },
      "agui": null
    },
    "recordVersion": "1.0.0",
    "status": "APPROVED"
  }
]

{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "batch_get_discoverable_registry_record",
    "arguments": {
      "recordIds": ["<RECORD_ID>"]
    }
  }
}

{
  "registryRecords": [
    {
      "registryArn": "arn:aws:agent-registry:us-east-1:<ACCOUNT_ID>:registry/<REGISTRY_ID>",
      "recordArn": "arn:aws:agent-registry:us-east-1:<ACCOUNT_ID>:registry/<REGISTRY_ID>/record/<RECORD_ID>",
      "recordId": "<RECORD_ID>",
      "name": "calculation-http-agent",
      "description": "Adds two numbers using the add_numbers tool. Hosted on AgentCore Runtime. POST JSON with a prompt string; returns a text/event-stream response. Authentication: AWS IAM (SigV4). Owner: demo team.",
      "displayName": "Calculation HTTP Agent",
      "recordType": "AGENT",
      "descriptors": {
        "mcpServer": null,
        "a2aAgentCard": null,
        "agentSkillsDefinition": null,
        "custom": null,
        "http": {
          "source": {
            "fromUrl": {
              "url": "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-east-1%3A<ACCOUNT_ID>%3Aruntime%2F<RUNTIME_ID>/invocations?qualifier=DEFAULT"
            }
          }
        },
        "agui": null
      },
      "recordVersion": "1.0.0",
      "status": "APPROVED",
      "createdAt": "2026-09-15T04:48:58.608183099Z",
      "updatedAt": "2026-09-15T04:49:05.596253649Z"
    }
  ],
  "errors": []
}

{
  "endpoint": "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/<URLエンコードしたRuntime ARN>/invocations?qualifier=DEFAULT",
  "authentication": {
    "type": "AWS_IAM",
    "signingService": "bedrock-agentcore"
  },
  "request": {
    "method": "POST",
    "contentType": "application/json",
    "example": {"prompt": "Use add_numbers to calculate 2 + 3."}
  },
  "response": {"contentType": "text/event-stream"},
  "owner": "demo team"
}