エージェントプラグインマーケットプレイスを構築する
ツールを一元管理するストアで、エージェントプラグインの公開、評価、配布を行います。
「エージェントプラグインマーケットプレイスを構築する」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
エージェントプラグインマーケットプレイスとは
プラグインマーケットプレイスは、開発者が共有可能なツールを公開し、他の開発者が検索して見つけ、エージェントシステムが自動的にインストールできる集中管理型のレジストリです。エージェントツール向けのnpmだと考えてください。一度公開すれば、どこでも利用できます。
マーケットプレイスのアーキテクチャ
マーケットプレイスには、主に4つのサービスがあります。Registry(プラグインのメタデータをデータベースに保存)、Storage(プラグインのアーカイブをオブジェクトストレージに保存)、Search(全文インデックス)、Security(パッケージへの署名と検証)です。すべてREST API経由で公開されます。
# 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}')プラグインの公開
公開エンドポイントは、プラグインアーカイブを検証し、マニフェストを確認し、公開者の身元をAPIキーで検証し、既知の悪意あるパターンをスキャンして、パッケージに署名したうえで保存します。公開処理はアトミックです。アップロード全体が成功するか、何も保存されないかのどちらかです。
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)}')検索によるプラグインの発見
検索APIは、フリーテキストのクエリとタグフィルターを受け付けます。重み付けスコアを使用して結果を順位付けします。完全な名前一致が最も高く評価され、次にタグの一致、最後に説明文のキーワード一致が続きます。
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評価とレビューのシステム
評価システムは、開発者が高品質なプラグインを選ぶのに役立ちます。1~5個の星による評価と、任意のテキストレビューを実装します。評価データの集計値(平均値、件数)はレジストリに保存し、検索結果で返します。
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')バージョンのYank(非推奨化)
公開済みのバージョンに重大なバグやセキュリティ脆弱性がある場合は、そのバージョンをyankします。レジストリ上で非表示にすることで、新しいインストールでは使用できなくしますが、既存のインストールは動作し続けます。削除すると既存のエージェントが動かなくなるため、Yankのほうが安全です。
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"')マーケットプレイスCLIツール
コマンドラインインターフェースを使うと、ターミナルやCIパイプラインからマーケットプレイスを簡単に利用できます。marketplace install、marketplace search、marketplace publish、marketplace listの各コマンドを実装します。
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}')理解度チェック
プラグインのバージョンをyankすることと、削除することの違いは何ですか。
まとめ:エージェントプラグインマーケットプレイスの構築
このレッスンを完了しました。おめでとうございます。主なポイント:
- 公開パイプライン:アーカイブを検証 → 署名を確認 → スキャン → 保存とインデックス化
- パッケージの署名:HMAC署名によって完全性を保証し、クライアントはインストール前に検証します
- 検索:タグフィルタリングと関連度ランキングに対応した全文検索
- インストール:ダウンロード → 検証 → 展開 → 検証 → 登録(失敗時はロールバック)
- 評価:1~5個の星とテキストレビューをレジストリで集計します
- Yank:既存のインストールを壊さずに、新しいインストールから非表示にします
最終コースは「自律システムへの道」です。
よくある質問
「エージェントプラグインマーケットプレイスを構築する」レッスンは無料ですか?
はい。「エージェントプラグインマーケットプレイスを構築する」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し(組み込みコードエディタと24時間対応のAIチューター)、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「エージェントプラグインマーケットプレイスを構築する」で何を学びますか?
ツールを一元管理するストアで、エージェントプラグインの公開、評価、配布を行います。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか?
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「エージェントプラグインマーケットプレイスを構築する」レッスンにはどのくらい時間がかかりますか?
ほとんどのCoddyKitレッスンは約5~10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか?
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- 共有可能なエージェントツールの設計
- プラグインの検出と登録
- ツールのバージョン管理と互換性
- エージェントプラグインマーケットプレイスを構築する