🤖 知识工作者的瑞士军刀：Anthropic 开源插件仓库深度解析

# plugin.yaml 插件清单示例
name: code-reviewer
version: 1.0.0
description: 自动审查 Pull Request 中的代码质量
triggers:
  - event: pull_request.created
  - event: pull_request.updated
actions:
  - name: analyze_code
    description: 分析代码并生成审查报告
    input:
      type: object
      properties:
        repo_url:
          type: string
          description: 仓库 URL
        pr_number:
          type: integer
    output:
      type: object
      properties:
        score:
          type: integer
          description: 代码质量评分 (0-100)
        suggestions:
          type: array
          items:
            type: string

{
  "jsonrpc": "2.0",
  "method": "execute_action",
  "params": {
    "plugin_id": "code-reviewer",
    "action": "analyze_code",
    "arguments": {
      "repo_url": "https://github.com/example/my-app",
      "pr_number": 42
    }
  },
  "id": 1
}

# 简化的插件加载器核心逻辑
import os
import json
import subprocess
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class PluginWatcher(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith('plugin.yaml'):
            plugin_config = self.load_manifest(event.src_path)
            self.register_plugin(plugin_config)
    
    def register_plugin(self, config):
        # 在独立的沙箱进程中启动插件
        process = subprocess.Popen(
            ['python', 'plugin_runner.py', json.dumps(config)],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE
        )
        self.active_plugins[config['name']] = process
        print(f"[Plugin Loader] 已加载插件: {config['name']} v{config['version']}")

# 插件上下文对象示例
class PluginContext:
    def __init__(self):
        self.current_document = None  # 当前打开的文档
        self.project_root = None      # 工作区根目录
        self.user_preferences = {}    # 用户偏好设置
        self.conversation_history = [] # 对话历史
    
    def get_file_content(self, filepath):
        """获取工作区中的文件内容"""
        full_path = os.path.join(self.project_root, filepath)
        if os.path.exists(full_path):
            with open(full_path, 'r', encoding='utf-8') as f:
                return f.read()
        return None

# 用户输入
"帮我总结这份 API 文档，提取所有 Endpoint 和请求参数"

# 插件处理流程
1. 读取 PDF 文件 → 2. OCR 识别（如果是扫描件）→ 3. 分段解析 → 4. 提取结构化数据
→ 5. 生成 Markdown 表格

# 输出示例
| Endpoint | Method | 参数 | 描述 |
|----------|--------|------|------|
| /api/v1/users | GET | page, limit | 获取用户列表 |
| /api/v1/users/:id | PUT | name, email | 更新用户信息 |

# 插件检测到的常见问题示例
def get_user_data(user_id):
    # 问题1：缺少输入验证
    # 问题2：直接拼接 SQL（SQL注入风险）
    query = f"SELECT * FROM users WHERE id = {user_id}"
    # 问题3：没有使用索引，全表扫描
    return database.execute(query)

# 安装脚手架
pip install anthropic-plugin-cli

# 创建新插件
create-plugin --name my-awesome-plugin --language python

# 进入插件目录
cd my-awesome-plugin

# 启动开发服务器（自动热重载）
plugin dev

my-awesome-plugin/
├── plugin.yaml          # 插件清单
├── src/
│   ├── __init__.py
│   ├── handler.py       # 主逻辑
│   └── utils.py         # 工具函数
├── tests/
│   └── test_handler.py
├── requirements.txt     # 依赖
└── README.md

# src/handler.py
import httpx
from plugin_sdk import PluginBase, Action

class WeatherPlugin(PluginBase):
    @Action(name="get_weather", description="获取指定城市的天气信息")
    async def get_weather(self, city: str) -> dict:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"https://api.weather.com/v1/{city}",
                params={"key": self.config["api_key"]}
            )
            data = response.json()
            return {
                "temperature": data["main"]["temp"],
                "humidity": data["main"]["humidity"],
                "description": data["weather"][0]["description"]
            }