0Pricing
AI Agents · 课时

构建智能体插件市场

集中式工具商店：发布、评分和分发智能体插件

构建智能体插件市场 是 CoddyKit 上的免费 AI Agents 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

什么是智能体插件市场

插件市场是一个集中式注册中心，开发者可以在其中发布可共享的工具，其他开发者可以通过搜索发现这些工具，智能体系统则可以自动安装它们。您可以把它看作面向智能体工具的包管理器：发布一次，随处使用。

市场架构

市场包含四项主要服务：注册表（将插件元数据存储在数据库中）、存储（将插件归档存储在对象存储中）、搜索（全文索引）和安全（对软件包进行签名和验证）。所有服务均通过表述性状态转移应用程序接口对外提供。

# 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

评分与评论系统

评分系统可以帮助开发者选择高质量的插件。请实现 1–5 星评分，并支持可选的文字评论。聚合后的评分数据（mean、数量）存储在注册表中，并在搜索结果中返回。

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"')

市场 CLI 工具

命令行界面让您可以方便地从终端或持续集成流程中使用市场。请实现 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}')

知识检查

撤回插件版本与删除插件版本有什么区别？

回顾：构建智能体插件市场

恭喜您完成本课！要点如下：

  • 发布流程：验证归档 → 验证签名 → 扫描 → 存储并建立索引
  • 软件包签名：基于哈希的消息认证码签名可确保完整性；客户端在安装前进行验证
  • 搜索：支持标签筛选和相关性排序的全文搜索
  • 安装：下载 → 验证 → 解压 → 验证 → 注册（失败时回滚）
  • 评分：1–5 星评分和文字评论在注册表中汇总
  • 撤回：对新安装隐藏版本，同时不影响现有安装

最后一门课程：通往自主系统之路。

常见问题解答

「构建智能体插件市场」课时是免费的吗？

是的 — 「构建智能体插件市场」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「构建智能体插件市场」这节课中我会学到什么？

集中式工具商店：发布、评分和分发智能体插件 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。

「构建智能体插件市场」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 设计可共享的智能体工具
  2. 插件发现与注册
  3. 工具版本管理与兼容性
  4. 构建智能体插件市场
← 返回 AI Agents