에이전트 플러그인 마켓플레이스 구축

에이전트 플러그인 마켓플레이스란 무엇입니까?

마켓플레이스 아키텍처

# Marketplace REST API endpoints:
API_ROUTES = {
    'POST /plugins': 'Publish a new plugin version',
    'GET /plugins': 'List all plugins (paginated)',
    'GET /plugins/search?q=weather': 'Search by keyword',
    'GET /plugins/{name}': 'Get plugin metadata',
    'GET /plugins/{name}/{version}': 'Get specific version metadata',
    'GET /plugins/{name}/{version}/download': 'Download plugin archive',
    'POST /plugins/{name}/{version}/reviews': 'Submit a rating/review',
    'GET /plugins/{name}/reviews': 'Get all reviews',
    'DELETE /plugins/{name}/{version}': 'Yank (hide) a version',
}

for route, description in API_ROUTES.items():
    print(f'{route}: {description}')

플러그인 게시

import hashlib
import hmac
import os

def publish_plugin(
    archive_path: str,
    api_key: str,
    registry_url: str = 'https://registry.agenttools.io'
) -> dict:
    import requests

    # Compute checksum for integrity
    with open(archive_path, 'rb') as f:
        data = f.read()
    sha256 = hashlib.sha256(data).hexdigest()

    # Sign the archive with the developer's API key
    signature = hmac.new(
        api_key.encode(), data, hashlib.sha256
    ).hexdigest()

    response = requests.post(
        f'{registry_url}/plugins',
        files={'archive': (os.path.basename(archive_path), data, 'application/zip')},
        headers={
            'X-API-Key': api_key,
            'X-Signature': signature,
            'X-Checksum-SHA256': sha256
        },
        timeout=60
    )
    response.raise_for_status()
    return response.json()

게시 시 서버 측 검증

import zipfile
import io
import json

MAX_ARCHIVE_SIZE_MB = 50

def validate_plugin_archive(archive_bytes: bytes) -> dict:
    errors = []

    # Size check
    size_mb = len(archive_bytes) / 1024 / 1024
    if size_mb > MAX_ARCHIVE_SIZE_MB:
        errors.append(f'Archive too large: {size_mb:.1f}MB (max {MAX_ARCHIVE_SIZE_MB}MB)')
        return {'valid': False, 'errors': errors}

    # Must be a valid zip
    try:
        zf = zipfile.ZipFile(io.BytesIO(archive_bytes))
    except zipfile.BadZipFile:
        errors.append('Not a valid zip archive')
        return {'valid': False, 'errors': errors}

    file_list = zf.namelist()

    # Must contain plugin.json
    if 'plugin.json' not in file_list:
        errors.append('Missing plugin.json in archive root')

    # Basic security scan: no .pyc files, no hidden dirs
    suspicious = [f for f in file_list if f.startswith('.') or '__pycache__' in f]
    if suspicious:
        errors.append(f'Suspicious files: {suspicious[:3]}')

    return {'valid': len(errors) == 0, 'errors': errors, 'files': len(file_list)}

if __name__ == '__main__':
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, 'w') as zf:
        zf.writestr('plugin.json', '{"name": "demo-plugin"}')
    archive_bytes = buf.getvalue()
    print('Validation result:', validate_plugin_archive(archive_bytes))

패키지 서명 및 검증

import hashlib
import hmac

def sign_package(archive_bytes: bytes, secret_key: str) -> str:
    return hmac.new(
        secret_key.encode(), archive_bytes, hashlib.sha256
    ).hexdigest()

def verify_package(
    archive_bytes: bytes,
    signature: str,
    secret_key: str
) -> bool:
    expected = sign_package(archive_bytes, secret_key)
    return hmac.compare_digest(expected, signature)

# Example:
package_data = b'fake plugin archive bytes'
developer_key = 'developer_secret_key_abc123'

sig = sign_package(package_data, developer_key)
print(f'Signature: {sig[:20]}...')

# Verify (registry or installer):
valid = verify_package(package_data, sig, developer_key)
print(f'Signature valid: {valid}')

# Tampered data:
tampered = b'tampered bytes'
print(f'Tampered valid: {verify_package(tampered, sig, developer_key)}')

검색을 통한 플러그인 찾기

import requests

def search_marketplace(
    query: str,
    tags: list = None,
    page: int = 0,
    per_page: int = 20,
    registry_url: str = 'https://registry.agenttools.io'
) -> dict:
    params = {
        'q': query,
        'page': page,
        'per_page': per_page
    }
    if tags:
        params['tags'] = ','.join(tags)

    response = requests.get(
        f'{registry_url}/plugins/search',
        params=params,
        timeout=10
    )
    response.raise_for_status()
    return response.json()

# Usage:
results = search_marketplace('weather forecast', tags=['iot', 'weather'])
# {
#   'total': 12,
#   'results': [
#     {'name': 'weather-tools', 'version': '2.1.0', 'score': 0.98, ...},
#     ...
#   ]
# }
print(f'Total results: {results.get("total", 0)}')

플러그인 설치

import requests
import zipfile
import io
import shutil
import os

def install_plugin(
    plugin_name: str,
    version: str = 'latest',
    plugins_dir: str = 'plugins',
    registry_url: str = 'https://registry.agenttools.io'
) -> bool:
    # Step 1: Get metadata including expected signature
    meta = requests.get(
        f'{registry_url}/plugins/{plugin_name}/{version}',
        timeout=10
    ).json()

    # Step 2: Download archive
    archive_resp = requests.get(
        meta['download_url'], timeout=60
    )
    archive_bytes = archive_resp.content

    # Step 3: Verify signature
    if not verify_package(archive_bytes, meta['signature'], meta['public_key']):
        print('Signature verification FAILED — aborting install')
        return False

    # Step 4: Extract
    install_path = os.path.join(plugins_dir, plugin_name)
    if os.path.exists(install_path):
        shutil.rmtree(install_path)  # remove old version
    with zipfile.ZipFile(io.BytesIO(archive_bytes)) as zf:
        zf.extractall(install_path)

    print(f'Installed {plugin_name} v{version} to {install_path}')
    return True

평가 및 리뷰 시스템

import requests

def submit_review(
    plugin_name: str,
    stars: int,
    review_text: str = '',
    api_key: str = '',
    registry_url: str = 'https://registry.agenttools.io'
) -> dict:
    if not 1 <= stars <= 5:
        raise ValueError('Stars must be 1-5')
    payload = {
        'stars': stars,
        'review': review_text[:500]  # truncate to max length
    }
    response = requests.post(
        f'{registry_url}/plugins/{plugin_name}/reviews',
        json=payload,
        headers={'X-API-Key': api_key},
        timeout=10
    )
    response.raise_for_status()
    return response.json()

def get_reviews(plugin_name: str, registry_url: str = 'https://registry.agenttools.io') -> dict:
    response = requests.get(
        f'{registry_url}/plugins/{plugin_name}/reviews',
        timeout=10
    )
    return response.json()

# Example response structure:
# {'average_stars': 4.3, 'total_reviews': 47,
#  'reviews': [{'stars': 5, 'review': 'Works great!', 'date': '...'}]}
print('Review system ready')

버전 얀크(사용 중단)

import requests

def yank_version(
    plugin_name: str,
    version: str,
    reason: str,
    api_key: str,
    registry_url: str = 'https://registry.agenttools.io'
) -> dict:
    response = requests.delete(
        f'{registry_url}/plugins/{plugin_name}/{version}',
        json={'reason': reason},
        headers={'X-API-Key': api_key},
        timeout=10
    )
    response.raise_for_status()
    return response.json()

# Yanked versions:
# - Still downloadable by users who have them pinned
# - Not returned in 'latest' queries
# - Search results show a 'yanked' badge with reason
# - Install of yanked version requires explicit --allow-yanked flag

print('Yanking removes a version from new installs without breaking existing ones.')
print('Always provide a reason: "Security vulnerability in API key handling"')

마켓플레이스 명령줄 도구

import argparse
import sys

def main():
    parser = argparse.ArgumentParser(description='Agent Tool Marketplace CLI')
    subparsers = parser.add_subparsers(dest='command')

    # Search
    search_parser = subparsers.add_parser('search', help='Search plugins')
    search_parser.add_argument('query', help='Search query')
    search_parser.add_argument('--tag', action='append', dest='tags')

    # Install
    install_parser = subparsers.add_parser('install', help='Install a plugin')
    install_parser.add_argument('plugin', help='Plugin name')
    install_parser.add_argument('--version', default='latest')
    install_parser.add_argument('--plugins-dir', default='plugins')

    # Publish
    publish_parser = subparsers.add_parser('publish', help='Publish plugin')
    publish_parser.add_argument('archive', help='Path to .zip archive')
    publish_parser.add_argument('--api-key', required=True)

    args = parser.parse_args()
    if args.command == 'search':
        results = search_marketplace(args.query, tags=args.tags or [])
        for r in results.get('results', []):
            print(f'{r["name"]} v{r["version"]} ({r.get("average_stars", "?")} stars)')
    elif args.command == 'install':
        install_plugin(args.plugin, args.version, args.plugins_dir)
    elif args.command == 'publish':
        result = publish_plugin(args.archive, args.api_key)
        print('Published:', result)

마켓플레이스 거버넌스

MARKETPLACE_POLICIES = {
    'namespace_policy': (
        'Plugin names must be unique across the registry. '
        'Names are claimed on first publish. '
        'Transfers require proof of original authorship.'
    ),
    'security_scanning': (
        'All uploads are scanned with bandit (Python security linter). '
        'Critical vulnerabilities block publish. '
        'High vulnerabilities generate a warning visible to installers.'
    ),
    'rate_limits': {
        'publish_per_hour': 10,
        'search_per_minute': 60,
        'install_per_minute': 30
    },
    'takedown_process': (
        'Maintainers can yank versions at any time. '
        'Legal/security takedowns processed within 48 hours. '
        'Appeals via security@agenttools.io'
    )
}

for policy, detail in MARKETPLACE_POLICIES.items():
    if isinstance(detail, str):
        print(f'{policy}: {detail[:80]}...')
    else:
        print(f'{policy}: {detail}')

지식 확인

복습: 에이전트 플러그인 마켓플레이스 구축