插件名称: 企业客户查询服务
域名: https://api.crm.example.com
描述: |
  提供企业客户信息查询服务，支持：
  - 客户基本信息查询（姓名、电话、邮箱）
  - 订单历史查询
  - 客户等级与标签查询
  
  使用前提：
  - 已配置API认证密钥
  - 已开通对应接口权限

# ✅ 好的描述示例
工具名称: get_weather
描述: |
  根据城市名称查询实时天气信息，返回温度、湿度、风力、空气质量等数据。
  适用于用户询问天气、出行建议、穿衣指数等场景。
  
  参数示例：city="北京" 返回北京当前天气
  参数示例：city="Shanghai" 返回上海当前天气（英文城市名也支持）

# ❌ 不好的描述示例
工具名称: get_weather
描述: 获取天气

# 查询参数配置
查询参数:
  - name: city
    type: string
    required: true
    description: |
      城市名称，支持以下格式：
      - 中文城市名：北京、上海、深圳
      - 英文城市名：beijing、shanghai
      - 城市拼音：beijing
      
  - name: language
    type: string
    required: false
    default: zh
    description: 返回语言，支持zh（中文）、en（英文）
    
  - name: days
    type: integer
    required: false
    default: 1
    description: 预报天数，1-7天，默认1天

# Header参数配置
Header参数:
  - name: Authorization
    type: string
    required: true
    description: Bearer Token认证，格式：Bearer {token}
    
  - name: X-Request-ID
    type: string
    required: false
    description: 请求唯一标识，用于追踪和排错

# Body参数配置（POST请求）
Body参数:
  - name: query
    type: string
    required: true
    description: 待查询的文本内容
    example: "Hello World"
    
  - name: source_lang
    type: string
    required: true
    description: 源语言，如zh、en、ja
    example: "en"
    
  - name: target_lang
    type: string
    required: true
    description: 目标语言，如zh、en、ja
    example: "zh"

认证类型: Header
Header名: X-API-Key
Header值: your_api_key_here

认证类型: Query参数
参数名: api_key
参数值: your_api_key_here

认证类型: Header
Header名: Authorization
Header值: Bearer your_token_here

认证类型: 动态参数
参数配置:
  appid: your_appid
  salt: "{salt}"  # 随机数
  sign: "{sign}"  # 签名，由算法生成

import hashlib

def make_sign(query, appid, salt, secret_key):
    """生成签名"""
    sign_str = f"{appid}{query}{salt}{secret_key}"
    return hashlib.md5(sign_str.encode()).hexdigest()

# 使用
salt = str(int(time.time()))
sign = make_sign("Hello", "your_appid", salt, "your_secret_key")

认证类型: OAuth 2.0
授权URL: https://oauth.example.com/authorize
令牌URL: https://oauth.example.com/token
客户端ID: your_client_id
客户端密钥: your_client_secret
作用域: read write

// API返回
{
  "temperature": 25,
  "humidity": 60,
  "weather": "晴"
}

// 映射配置
{
  "temperature": "temperature",
  "humidity": "humidity",
  "weather": "weather"
}

// API返回
{
  "data": {
    "current": {
      "temp": 25,
      "condition": "晴"
    }
  },
  "code": 200,
  "message": "success"
}

// 映射配置
{
  "temperature": "data.current.temp",
  "weather": "data.current.condition",
  "status_code": "code",
  "message": "message"
}

// API返回
{
  "results": [
    {"name": "张三", "age": 30},
    {"name": "李四", "age": 25}
  ]
}

// 映射配置
{
  "users": "results"  // 保留列表结构
}

# ✅ 推荐：清晰、符合规范
city: "城市名称"
start_date: "开始日期，格式YYYY-MM-DD"
user_email: "用户邮箱地址"
page_size: "每页数量，默认20"

# ❌ 不推荐：过于简略
c: "城市"
s: "开始日期"
e: "邮箱"
p: "页数"

# 整数类型 - 用于数值数据
temperature: integer  # 温度：25
count: integer       # 数量：10

# 数字类型 - 用于浮点数
price: number        # 价格：19.99
latitude: number     # 纬度：39.9042

# 字符串类型 - 用于文本
city: string        # 城市名
content: string     # 内容

# 布尔类型 - 用于开关/状态
is_vip: boolean     # 是否VIP
enable_cache: boolean # 是否启用缓存

# 数组类型 - 用于列表数据
tags: array         # 标签列表
emails: array       # 邮箱列表

# 对象类型 - 用于复杂结构
user_info: object   # 用户信息对象
address: object     # 地址对象

# 必填参数（没有默认值）
required_params:
  - name: city
    type: string
    required: true
    description: "城市名称，必填，不支持空值"
    
  - name: api_key
    type: string
    required: true
    description: "API密钥，必填"

# 可选参数（带默认值）
optional_params:
  - name: language
    type: string
    required: false
    default: "zh"
    description: "返回语言，默认中文(zh)，可选英文(en)"
    
  - name: timeout
    type: integer
    required: false
    default: 30
    description: "超时时间（秒），默认30秒，最大120秒"

# ✅ 优秀的描述示例
city:
  description: |
    城市名称，支持以下格式：
    - 中文城市名：北京、上海、深圳、广州
    - 英文城市名：beijing、shanghai、guangzhou
    - 城市拼音：beijing、shanghai
    
    注意：
    1. 县级市请使用所属城市+县名，如"北京市昌平区"
    2. 国外城市请使用英文名
    3. 不确定时建议使用中文全名

date_range:
  description: |
    日期范围，格式为"开始日期,结束日期"
    - 示例：2024-01-01,2024-01-07
    - 支持相对日期：如"today,7days_later"
    - 最大范围：90天

# ❌ 简单的描述示例
city:
  description: "城市名称"

// API返回
{
  "temperature": 25,
  "humidity": 60,
  "weather": "晴",
  "wind": "南风3级"
}

// 映射配置
{
  "temperature": "temperature",
  "humidity": "humidity",
  "weather": "weather",
  "wind": "wind"
}

// API返回
{
  "status": "success",
  "data": {
    "location": {
      "city": "北京",
      "district": "朝阳区"
    },
    "now": {
      "temp": 26,
      "feels_like": 28,
      "condition": "多云"
    }
  }
}

// 映射配置
{
  "status": "status",
  "city": "data.location.city",
  "district": "data.location.district",
  "temperature": "data.now.temp",
  "feels_like": "data.now.feels_like",
  "weather": "data.now.condition"
}

// 成功响应
{
  "code": 200,
  "data": {
    "result": "翻译结果"
  }
}

// 失败响应
{
  "code": 400,
  "error": "Invalid language code"
}

// 映射配置 - 统一处理
{
  "code": "code",
  "result": "data.result",  // 成功时有效
  "error": "error"          // 失败时有效
}

// API返回
{
  "total": 100,
  "page": 1,
  "page_size": 10,
  "items": [
    {"id": 1, "name": "产品A", "price": 99},
    {"id": 2, "name": "产品B", "price": 199}
  ]
}

// 映射配置
{
  "total": "total",
  "page": "page",
  "page_size": "page_size",
  "products": "items"  // 保留完整列表
}

认证类型: Header
Header名: X-API-Key
Header值: ${API_KEY}  # 变量形式，便于管理

认证类型: Query参数
参数名: api_key
参数值: ${API_KEY}

认证类型: Header
Header名: Authorization
Header值: Bearer ${ACCESS_TOKEN}

# 签名算法示例（以百度翻译为例）
import hashlib
import time
import random

def generate_baidu_sign(query, appid, secret_key):
    """
    生成百度翻译API签名
    
    签名规则：MD5(appid + q + salt + 密钥)
    """
    salt = str(random.randint(32768, 65536))
    sign_str = f"{appid}{query}{salt}{secret_key}"
    sign = hashlib.md5(sign_str.encode()).hexdigest()
    
    return {
        'appid': appid,
        'q': query,
        'salt': salt,
        'sign': sign
    }

# 使用示例
params = generate_baidu_sign(
    query="Hello",
    appid="your_appid",
    secret_key="your_secret_key"
)
# 返回：{'appid': '...', 'q': 'Hello', 'salt': '12345', 'sign': 'abcdef123456'}

# 数据处理
pandas>=1.3.0
numpy>=1.20.0

# HTTP请求（虽然平台已有requests，但httpx支持异步）
httpx>=0.24.0

# JSON处理（orjson性能更好）
orjson>=3.8.0

# 数据验证
pydantic>=2.0.0

# 日期时间处理
python-dateutil>=2.8.0

# 加密解密
cryptography>=41.0.0

插件节点输入配置:
  插件名称: 天气服务
  工具名称: get_current_weather
  
  输入参数映射:
    city: "${用户输入节点.city}"        # 直接引用上游输出
    date: "${日期处理节点.formatted}"   # 经过处理的数据
    language: "zh"                      # 固定值

插件节点输出配置:
  提取字段:
    temperature: "实时温度(℃)"
    humidity: "湿度(%)"
    weather: "天气状况"
    wind_speed: "风速"
    
  输出变量:
    weather_result: "${插件节点}"

工作流设计:
  节点1: 语言检测
    类型: JavaScript
    代码: |
      function detectLanguage(text) {
        // 简化的语言检测逻辑
        const chineseRegex = /[\u4e00-\u9fa5]/;
        if (chineseRegex.test(text)) return 'zh';
        const japaneseRegex = /[\u3040-\u309f\u30a0-\u30ff]/;
        if (japaneseRegex.test(text)) return 'ja';
        return 'en';
      }
      return { language: detectLanguage(input.text) };

  节点2: 条件分支
    类型: 条件分支
    条件:
      - IF language == "zh" → 调用中译英插件
      - IF language == "en" → 调用英译中插件
      - IF language == "ja" → 调用日译中插件
      - ELSE → 调用通用翻译插件

city: "${输入节点.city}"

date: "${日期节点.iso_format}"

language: "zh"
format: "json"

end_date: "${开始日期节点.date} + 7days"  # 需要JS节点处理

temperature: "${天气插件.temperature}"

weather_data:
  temperature: "${天气插件.temperature}"
  humidity: "${天气插件.humidity}"
  weather: "${天气插件.weather}"

raw_result: "${天气插件.*}"  # 传递整个响应对象

错误处理配置:
  超时设置: 30秒
  
  重试策略:
    最大重试次数: 3
    重试间隔: 2秒
    指数退避: 是  # 第1次2秒，第2次4秒，第3次8秒
  
  错误兜底:
    - 返回默认值: "暂时无法获取天气信息"
    - 降级处理: "建议您稍后重试"
    - 记录日志: 是
    - 发送告警: 否

# 调用聚合数据多个接口
插件域名: https://apis.juhe.cn

# 工具1：天气查询
GET /weather?city={city}
描述: 查询实时天气

# 工具2：空气质量查询
GET /air-quality?city={city}
描述: 查询空气质量

# 工具3：生活指数
GET /life-index?city={city}
描述: 查询穿衣指数、紫外线指数等

# ❌ 错误：一个插件包含不同域名的工具
工具1: https://api.weather.com/now      # 天气 - 域名1
工具2: https://api.news.com/latest       # 新闻 - 域名2 ❌

# ✅ 正确：拆分为多个插件
插件1（天气服务）: 
  域名: https://api.weather.com
  工具: /now, /forecast
  
插件2（新闻服务）: 
  域名: https://api.news.com
  工具: /latest, /hot

def handle_api_response(response):
    """处理API响应，根据状态码和业务码进行不同处理"""
    
    # HTTP状态码检查
    if response.status_code == 200:
        data = response.json()
        
        # 业务状态码检查
        if data.get('error_code') == 0:
            return {
                'success': True,
                'data': data.get('result')
            }
        else:
            # 业务错误
            return {
                'success': False,
                'error': f"API业务错误: {data.get('reason', '未知错误')}",
                'error_code': data.get('error_code')
            }
    
    # HTTP客户端错误
    elif response.status_code == 400:
        return {'success': False, 'error': '请求参数错误'}
    elif response.status_code == 401:
        return {'success': False, 'error': '认证失败，请检查API密钥'}
    elif response.status_code == 403:
        return {'success': False, 'error': '权限不足'}
    elif response.status_code == 404:
        return {'success': False, 'error': '资源不存在'}
    elif response.status_code == 429:
        return {'success': False, 'error': '请求过于频繁，请稍后重试'}
    
    # HTTP服务器错误
    elif response.status_code >= 500:
        return {'success': False, 'error': '服务器繁忙，请稍后重试'}
    
    else:
        return {'success': False, 'error': f'HTTP错误: {response.status_code}'}

import time
from requests.exceptions import Timeout, ConnectionError, HTTPError

def call_api_with_retry(url, params, max_retries=3, timeout=10):
    """
    带重试机制的API调用
    
    参数:
        url: API地址
        params: 请求参数
        max_retries: 最大重试次数
        timeout: 单次请求超时时间（秒）
    
    返回:
        API响应数据
    """
    for attempt in range(max_retries):
        try:
            response = requests.get(url, params=params, timeout=timeout)
            response.raise_for_status()
            return response.json()
            
        except Timeout:
            # 超时重试
            if attempt < max_retries - 1:
                wait_time = (attempt + 1) * 2  # 指数退避
                print(f"请求超时，{wait_time}秒后重试...")
                time.sleep(wait_time)
            else:
                raise Exception(f"API调用超时，已重试{max_retries}次")
                
        except ConnectionError:
            # 连接错误重试
            if attempt < max_retries - 1:
                wait_time = (attempt + 1) * 2
                print(f"连接失败，{wait_time}秒后重试...")
                time.sleep(wait_time)
            else:
                raise Exception(f"网络连接失败，已重试{max_retries}次")
                
        except HTTPError as e:
            # HTTP错误不重试，直接抛出
            raise Exception(f"HTTP错误: {e}")
    
    return None

def validate_response(data, required_fields):
    """
    校验返回数据的完整性和合法性
    
    参数:
        data: API返回的数据
        required_fields: 必填字段列表
    
    返回:
        校验通过返回True，失败抛出异常
    """
    if not data:
        raise ValueError("响应数据为空")
    
    # 检查必填字段
    for field in required_fields:
        if field not in data:
            raise ValueError(f"缺少必填字段: {field}")
    
    # 类型校验
    if 'temperature' in data:
        if not isinstance(data['temperature'], (int, float)):
            raise TypeError(f"温度字段类型错误，期望数字，实际{type(data['temperature'])}")
    
    if 'city' in data:
        if not isinstance(data['city'], str):
            raise TypeError(f"城市名字段类型错误，期望字符串")
    
    # 范围校验
    if 'temperature' in data:
        temp = data['temperature']
        if not -50 <= temp <= 60:  # 合理温度范围
            raise ValueError(f"温度数据异常: {temp}℃，超出合理范围")
    
    return True

# ❌ 不推荐：逐条查询
def query_users_individually(user_ids):
    results = []
    for user_id in user_ids:
        result = call_api(f"/user/{user_id}")
        results.append(result)
    return results

# ✅ 推荐：批量查询（如果API支持）
def query_users_batch(user_ids):
    # 一次性查询多个用户
    result = call_api("/users/batch", {"ids": ",".join(user_ids)})
    return result.get('users', [])

import time
from functools import lru_cache

# 简单的内存缓存
cache = {}
CACHE_TTL = 3600  # 缓存有效期：1小时

def get_weather_cached(city):
    """带TTL的天气缓存"""
    current_time = time.time()
    
    # 检查缓存
    if city in cache:
        cached_data, cached_time = cache[city]
        if current_time - cached_time < CACHE_TTL:
            print(f"命中缓存: {city}")
            return cached_data
    
    # 缓存未命中，调用API
    print(f"调用API查询: {city}")
    data = call_weather_api(city)
    
    # 更新缓存
    cache[city] = (data, current_time)
    return data

import asyncio
import httpx

async def concurrent_api_calls(urls):
    """并发API调用，显著提升效率"""
    async with httpx.AsyncClient() as client:
        # 创建所有任务
        tasks = [client.get(url) for url in urls]
        
        # 并发执行
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 处理结果
        results = []
        for i, response in enumerate(responses):
            if isinstance(response, Exception):
                print(f"请求{i}失败: {response}")
                results.append(None)
            else:
                results.append(response.json())
        
        return results

# 使用示例
async def main():
    urls = [
        "https://api.example.com/city/beijing",
        "https://api.example.com/city/shanghai",
        "https://api.example.com/city/guangzhou"
    ]
    results = await concurrent_api_calls(urls)

# 天气服务插件 v1.2.0

## 插件简介

提供实时天气查询服务，支持国内外3000+城市。

| 项目 | 说明 |
|-----|------|
| 版本 | v1.2.0 |
| 更新日期 | 2024-01-15 |
| 维护者 | xxx@example.com |

## 功能列表

- [x] 实时温度查询
- [x] 湿度查询
- [x] 风力查询
- [x] 空气质量查询
- [x] 未来3天预报
- [ ] 未来7天预报（开发中）

## 使用限制

| 限制项 | 限制值 |
|-------|--------|
| QPS限制 | 10次/秒 |
| 每日调用次数 | 10000次/天 |
| 支持城市 | 3000+ |

## 认证方式

API Key认证，需要在扣子平台配置X-API-Key

## 工具列表

### get_current_weather
获取当前天气

**输入参数**：

| 参数 | 类型 | 必填 | 默认值 | 说明 |
|-----|------|-----|-------|------|
| city | string | 是 | - | 城市名称 |
| language | string | 否 | zh | 返回语言(zh/en) |

**输出参数**：

| 参数 | 类型 | 说明 |
|-----|------|------|
| temperature | integer | 温度（摄氏度） |
| humidity | integer | 湿度（%） |
| weather | string | 天气状况 |
| wind_speed | string | 风速 |

**调用示例**：

# ✅ 正确配置
插件域名: https://api.example.com

工具1路径: /v1/weather/current       # 相对路径
工具2路径: /v1/weather/forecast      # 相对路径

# ❌ 错误配置
工具1路径: https://api.example.com/v1/weather/current   # 完整URL
工具2路径: https://api.another.com/v1/weather/current    # 错误域名

# ✅ 修正方案
如果需要调用不同域名，创建多个插件：
插件1（天气服务）: 
  域名: https://api.weather.com
插件2（新闻服务）: 
  域名: https://api.news.com

# 调试：打印实际发送的认证头
def debug_auth():
    api_key = "your_api_key_here"
    headers = {
        "X-API-Key": api_key,
        "Content-Type": "application/json"
    }
    print(f"发送的Headers: {headers}")
    # 验证与API文档要求是否一致

# 检查Token是否过期
def is_token_valid(token_expiry):
    from datetime import datetime
    return datetime.now() < token_expiry

# 测试网络连通性
ping api.example.com
telnet api.example.com 443

# 测试DNS解析
nslookup api.example.com

插件配置:
  超时时间: 30秒  # 默认10秒，适当增加

# 减少不必要的数据请求
# 使用分页而非一次性获取全量数据
# 只请求需要的字段

# 优化前：获取所有字段
params = {"city": city}

# 优化后：只获取必要字段
params = {"city": city, "fields": "temperature,humidity,weather"}

import time
from collections import deque

class RateLimiter:
    """滑动窗口限流器"""
    
    def __init__(self, max_calls, period):
        self.max_calls = max_calls  # 最大调用次数
        self.period = period        # 时间窗口（秒）
        self.calls = deque()        # 调用时间记录
    
    def acquire(self):
        """获取调用许可，阻塞直到获取成功"""
        now = time.time()
        
        # 清理过期记录
        while self.calls and self.calls[0] < now - self.period:
            self.calls.popleft()
        
        # 检查是否超限
        if len(self.calls) >= self.max_calls:
            # 需要等待
            sleep_time = self.calls[0] + self.period - now
            if sleep_time > 0:
                print(f"限流中，等待{sleep_time:.2f}秒...")
                time.sleep(sleep_time)
        
        # 记录本次调用
        self.calls.append(time.time())

# 使用
limiter = RateLimiter(max_calls=50, period=1)  # 50 QPS

for item in items:
    limiter.acquire()  # 先获取许可
    result = call_api(item)  # 再调用

import queue
import threading

class RequestQueue:
    """请求队列，平滑请求峰值"""
    
    def __init__(self, rate_limit=50):
        self.queue = queue.Queue()
        self.rate_limit = rate_limit
        self.processing = False
    
    def add_request(self, request):
        self.queue.put(request)
        if not self.processing:
            self._process_queue()
    
    def _process_queue(self):
        self.processing = True
        while not self.queue.empty():
            limiter.acquire()
            request = self.queue.get()
            process(request)
        self.processing = False

def debug_response(response):
    print(f"状态码: {response.status_code}")
    print(f"响应头: {response.headers}")
    print(f"响应内容: {response.text}")  # 打印原始文本

import json

def analyze_response(data):
    if isinstance(data, str):
        data = json.loads(data)
    
    print(f"数据类型: {type(data)}")
    print(f"顶级Key: {data.keys() if isinstance(data, dict) else 'N/A'}")
    
    # 递归打印结构
    def print_structure(obj, indent=0):
        prefix = "  " * indent
        if isinstance(obj, dict):
            for k, v in list(obj.items())[:5]:  # 只打印前5个
                print(f"{prefix}{k}: ", end="")
                if isinstance(v, (dict, list)):
                    print(f"({type(v).__name__})")
                    print_structure(v, indent + 1)
                else:
                    print(f"{v}")
        elif isinstance(obj, list):
            print(f"List[{len(obj)}]")
            if obj:
                print_structure(obj[0], indent + 1)
    
    print_structure(data)

# 确保参数路径与实际数据匹配
# 注意大小写敏感性

# API返回
{"Temperature": 25}  # 注意大写T

# ❌ 错误的映射
temperature: "temperature"

# ✅ 正确的映射
temperature: "Temperature"

插件名称: 心知天气服务
域名: https://api.seniverse.com
描述: |
  提供实时天气、预报、空气质量查询服务
  
  支持功能：
  - 实时天气查询
  - 未来三天预报
  - 空气质量查询
  
  使用前提：已获取心知天气API Key

工具名称: get_current_weather
描述: |
  根据城市名称查询实时天气信息，返回温度、湿度、风力等数据。
  适用于用户询问天气、出行建议等场景。
  
  参数示例：city="北京"
  返回：北京当前天气、温度、湿度、风力等信息

请求方法: GET
请求路径: /v3/weather/now.json

查询参数:
  - name: location
    type: string
    required: true
    description: 城市名称，支持中英文、拼音
    
  - name: key
    type: string
    required: true
    description: API密钥
    
  - name: language
    type: string
    required: false
    default: zh-chs
    description: 返回语言，默认简体中文
    
  - name: unit
    type: string
    required: false
    default: c
    description: 温度单位，c=摄氏度，f=华氏度

响应参数映射:
  city_name: results[0].location.name
  country: results[0].location.country
  temperature: results[0].now.temperature
  humidity: results[0].now.humidity
  weather_text: results[0].now.text
  wind_direction: results[0].now.wind_direction
  wind_scale: results[0].now.wind_scale
  visibility: results[0].now.vis
  pressure: results[0].now.pressure

工具名称: get_weather_forecast
描述: |
  查询未来三天的天气预报
  
  参数示例：city="上海"
  返回：上海未来三天的天气、温度范围、降水概率等

请求方法: GET
请求路径: /v3/weather/daily.json

查询参数:
  - name: location
    type: string
    required: true
    description: 城市名称
    
  - name: key
    type: string
    required: true
    description: API密钥
    
  - name: language
    type: string
    required: false
    default: zh-chs
    
  - name: unit
    type: string
    required: false
    default: c
    
  - name: start
    type: integer
    required: false
    default: 0
    description: 预报开始天数，0=今天

响应参数映射:
  city_name: results[0].location.name
  forecast_data: results[0].daily
  # daily包含：date, text_day, text_night, high, low, rain_prob, wind_direction, wind_scale等

工具名称: get_air_quality
描述: |
  查询城市空气质量，包含AQI、PM2.5、PM10等指标
  
  参数示例：city="广州"
  返回：广州当前空气质量、AQI等级、首要污染物等

请求方法: GET
请求路径: /v3/air/now.json

查询参数:
  - name: location
    type: string
    required: true
    description: 城市名称
    
  - name: key
    type: string
    required: true
    description: API密钥

响应参数映射:
  city_name: results[0].location.name
  aqi: results[0].air_now.city.aqi
  aqi_level: results[0].air_now.city.level
  pm25: results[0].air_now.city.pm25
  pm10: results[0].air_now.city.pm10
  so2: results[0].air_now.city.so2
  no2: results[0].air_now.city.no2
  co: results[0].air_now.city.co
  o3: results[0].air_now.city.o3
  primary_pollutant: results[0].air_now.city.primary_pollutant

智能体配置:
  人设: |
    你是一个贴心的天气助手，可以查询天气和空气质量。
    当用户询问天气时，主动调用天气插件获取准确信息。
    回复要简洁友好，结合天气给出实用建议。
  
  插件:
    - 心知天气服务:
      - get_current_weather
      - get_weather_forecast
      - get_air_quality

插件名称: 数据库查询服务
域名: https://api.your-company.com

工具: query_customer
描述: |
  根据条件查询客户信息
  
  支持的查询条件：
  - customer_id: 客户ID（精确查询）
  - name: 客户名称（模糊查询）
  - phone: 手机号（精确查询）
  - vip_level: VIP等级（精确查询）
  
  返回客户的基本信息、联系方式、VIP等级等

请求方法: POST
请求路径: /v1/database/query

Body参数:
  - name: table
    type: string
    required: true
    description: 表名，如 customers, orders
    
  - name: conditions
    type: object
    required: false
    description: 查询条件，JSON对象格式
    example: {"vip_level": "gold", "status": "active"}
    
  - name: fields
    type: array
    required: false
    description: 返回字段，默认返回所有字段
    example: ["id", "name", "phone", "vip_level"]
    
  - name: limit
    type: integer
    required: false
    default: 100
    description: 返回记录数限制，最大1000
    
  - name: offset
    type: integer
    required: false
    default: 0
    description: 偏移量，用于分页

# Python Flask 服务端实现
from flask import Flask, request, jsonify
import pymysql

app = Flask(__name__)

@app.route('/v1/database/query', methods=['POST'])
def query_database():
    data = request.json
    
    table = data.get('table')
    conditions = data.get('conditions', {})
    fields = data.get('fields', '*')
    limit = min(data.get('limit', 100), 1000)
    offset = data.get('offset', 0)
    
    # 构建查询语句
    field_str = ', '.join(fields) if isinstance(fields, list) else '*'
    where_clauses = [f"{k} = %s" for k in conditions.keys()]
    where_str = ' AND '.join(where_clauses) if where_clauses else '1=1'
    
    sql = f"SELECT {field_str} FROM {table} WHERE {where_str} LIMIT {limit} OFFSET {offset}"
    
    try:
        with get_db_connection() as conn:
            with conn.cursor(pymysql.cursors.DictCursor) as cursor:
                cursor.execute(sql, list(conditions.values()))
                results = cursor.fetchall()
                
        return jsonify({
            'success': True,
            'data': results,
            'total': len(results)
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

工具: insert_record
描述: |
  向数据库插入记录
  
  支持的表：orders, logs, feedbacks
  
  返回新记录的ID

请求方法: POST
请求路径: /v1/database/insert

Body参数:
  - name: table
    type: string
    required: true
    description: 表名
    
  - name: data
    type: object
    required: true
    description: 要插入的数据，JSON对象
    example: {"customer_id": 123, "amount": 999, "status": "pending"}

工具: send_notification
描述: |
  发送通知消息到队列
  
  支持的通知类型：
  - email: 发送邮件
  - sms: 发送短信
  - webhook: 触发Webhook
  - dingtalk: 钉钉通知

请求方法: POST
请求路径: /v1/notify/send

Body参数:
  - name: type
    type: string
    required: true
    description: 通知类型
    enum: [email, sms, webhook, dingtalk]
    
  - name: recipient
    type: string
    required: true
    description: 接收者标识（邮箱/手机号/URL等）
    
  - name: template
    type: string
    required: true
    description: 通知模板ID
    
  - name: params
    type: object
    required: false
    description: 模板参数
    example: {"name": "张三", "order_id": "ORDER123"}

# 异步处理示例
import pika
import json

def process_notification(ch, method, properties, body):
    """消费通知消息"""
    message = json.loads(body)
    
    notification_type = message['type']
    recipient = message['recipient']
    content = message['content']
    
    try:
        if notification_type == 'email':
            send_email(recipient, content)
        elif notification_type == 'sms':
            send_sms(recipient, content)
        elif notification_type == 'webhook':
            send_webhook(recipient, content)
            
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f"处理失败: {e}")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

# 启动消费者
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.basic_consume(queue='notifications', on_message_callback=process_notification)
channel.start_consuming()

import logging
from datetime import datetime

class PluginLogger:
    """插件日志记录器"""
    
    def __init__(self, plugin_name):
        self.logger = logging.getLogger(plugin_name)
        self.logger.setLevel(logging.INFO)
        
        # 添加文件Handler
        fh = logging.FileHandler(f'/var/log/plugins/{plugin_name}.log')
        fh.setLevel(logging.INFO)
        
        # 添加控制台Handler
        ch = logging.StreamHandler()
        ch.setLevel(logging.DEBUG)
        
        # 格式化
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        fh.setFormatter(formatter)
        ch.setFormatter(formatter)
        
        self.logger.addHandler(fh)
        self.logger.addHandler(ch)
    
    def log_request(self, tool_name, params):
        """记录请求"""
        self.logger.info(f"请求开始 - 工具:{tool_name} - 参数:{params}")
    
    def log_response(self, tool_name, result, duration):
        """记录响应"""
        self.logger.info(
            f"请求完成 - 工具:{tool_name} - 耗时:{duration:.3f}秒 - "
            f"结果:{result.get('success', False)}"
        )
    
    def log_error(self, tool_name, error):
        """记录错误"""
        self.logger.error(f"请求失败 - 工具:{tool_name} - 错误:{error}")

import requests
from datetime import datetime, timedelta

class PluginAlerter:
    """插件告警器"""
    
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url
    
    def send_alert(self, level, message, details=None):
        """发送告警"""
        alert_data = {
            "msg_type": "interactive",
            "card": {
                "config": {"wide_screen_mode": True},
                "elements": [
                    {
                        "tag": "markdown",
                        "content": f"**⚠️ 插件告警 [{level}]**\n\n{message}"
                    },
                    {
                        "tag": "div",
                        "text": {
                            "tag": "lark_md",
                            "content": f"**详情**: {details or '无'}"
                        }
                    },
                    {
                        "tag": "div",
                        "text": {
                            "tag": "lark_md",
                            "content": f"**时间**: {datetime.now().isoformat()}"
                        }
                    }
                ]
            }
        }
        
        try:
            requests.post(self.webhook_url, json=alert_data)
        except Exception as e:
            print(f"发送告警失败: {e}")
    
    def check_and_alert(self, metrics):
        """检查指标并告警"""
        # 检查成功率
        if metrics['success_rate'] < 0.95:
            self.send_alert(
                "WARNING",
                f"插件成功率低于95%: {metrics['success_rate']:.2%}",
                f"调用量: {metrics['total_calls']}, 成功: {metrics['success_calls']}"
            )
        
        # 检查响应时间
        if metrics['avg_duration'] > 3.0:
            self.send_alert(
                "INFO",
                f"平均响应时间过长: {metrics['avg_duration']:.2f}秒",
                f"P99: {metrics['p99_duration']:.2f}秒"
            )

# 变更日志

## v1.2.0 (2024-01-15)

### 新增
- [新增] get_air_quality 空气质量查询工具
- [新增] 支持多语言返回 (zh/en/ja)

### 优化
- [优化] 天气数据缓存策略，命中率提升30%
- [优化] 错误处理逻辑，添加详细错误码

### 修复
- [修复] 城市名称包含空格时查询失败的问题
- [修复] 极端天气数据解析异常

---

## v1.1.0 (2023-12-20)

### 新增
- [新增] get_weather_forecast 天气预报工具
- [新增] 支持未来7天预报

### 优化
- [优化] API调用重试机制

# ❌ 不推荐：在代码中硬编码密钥
API_KEY = "sk-xxxxxx-secret-key"

# ✅ 推荐：从环境变量或密钥管理服务获取
import os

API_KEY = os.environ.get('WEATHER_API_KEY')
# 或使用密钥管理服务
API_KEY = secrets_manager.get_secret('weather-api-key')

def validate_input(params, schema):
    """输入参数验证"""
    errors = []
    
    for field, rules in schema.items():
        value = params.get(field)
        
        # 必填检查
        if rules.get('required') and not value:
            errors.append(f"必填字段缺失: {field}")
            continue
        
        # 类型检查
        if value and not isinstance(value, rules['type']):
            errors.append(
                f"字段类型错误: {field}, "
                f"期望{type_names[rules['type']]}, 实际{type(value).__name__}"
            )
        
        # 范围检查
        if 'min' in rules and value < rules['min']:
            errors.append(f"字段值过小: {field}, 最小值{rules['min']}")
        if 'max' in rules and value > rules['max']:
            errors.append(f"字段值过大: {field}, 最大值{rules['max']}")
        
        # 枚举检查
        if 'enum' in rules and value not in rules['enum']:
            errors.append(f"字段值非法: {field}, 可选值{rules['enum']}")
    
    if errors:
        raise ValueError('; '.join(errors))
    
    return True

# 使用示例
schema = {
    'city': {'required': True, 'type': str},
    'days': {'required': False, 'type': int, 'min': 1, 'max': 7},
    'format': {'required': False, 'type': str, 'enum': ['json', 'xml']}
}

validate_input({'city': '北京', 'days': 3, 'format': 'json'}, schema)

import re

def mask_sensitive_data(data):
    """脱敏处理"""
    if isinstance(data, dict):
        return {k: mask_sensitive_data(v) for k, v in data.items()}
    elif isinstance(data, list):
        return [mask_sensitive_data(item) for item in data]
    elif isinstance(data, str):
        # 手机号脱敏：138****5678
        data = re.sub(r'(\d{3})\d{4}(\d{4})', r'\1****\2', data)
        # 邮箱脱敏：t***@example.com
        data = re.sub(r'(\w)\w*(@\w+\.\w+)', r'\1***\2', data)
        # 身份证脱敏
        data = re.sub(r'(\d{4})\d{10}(\d{4})', r'\1**********\2', data)
    return data

# 日志记录时自动脱敏
def log_with_mask(data):
    """脱敏后记录日志"""
    masked_data = mask_sensitive_data(data)
    logger.info(f"请求数据: {masked_data}")

运行环境:
  - Python >= 3.8
  - Node.js >= 14 (可选)
  
Python包:
  - flask >= 2.0
  - gunicorn >= 20.0
  - requests >= 2.25
  - pyyaml >= 5.4
  - pymysql >= 0.9
  - redis >= 3.5
  
运维工具:
  - Docker >= 20.0
  - Prometheus (监控)
  - Grafana (可视化)
  - ELK Stack (日志)

FROM python:3.9-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 设置环境变量
ENV PYTHONUNBUFFERED=1
ENV FLASK_ENV=production

# 暴露端口
EXPOSE 5000

# 启动命令
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "4", "--timeout", "30", "app:app"]

version: '3.8'

services:
  plugin-api:
    build: .
    ports:
      - "5000:5000"
    environment:
      - FLASK_ENV=production
      - DATABASE_URL=${DATABASE_URL}
      - REDIS_URL=${REDIS_URL}
    depends_on:
      - redis
      - mysql
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  redis:
    image: redis:6-alpine
    volumes:
      - redis_data:/data
    restart: unless-stopped

  mysql:
    image: mysql:8
    environment:
      - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
      - MYSQL_DATABASE=plugin_db
    volumes:
      - mysql_data:/var/lib/mysql
    restart: unless-stopped

volumes:
  redis_data:
  mysql_data:

# .github/workflows/deploy.yml
name: Deploy Plugin API

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      
      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.9'
      
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-cov
      
      - name: Run tests
        run: pytest --cov=app tests/
      
      - name: Lint
        run: |
          pip install flake8
          flake8 app --count --select=E9,F63,F7,F82 --show-source --statistics

  deploy:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    
    steps:
      - name: Deploy to server
        uses: appleboy/ssh-action@master
        with:
          host: ${{ secrets.HOST }}
          username: ${{ secrets.USERNAME }}
          key: ${{ secrets.SSH_KEY }}
          script: |
            cd /app/plugin-api
            docker-compose pull
            docker-compose up -d
            docker-compose exec -T plugin-api python manage.py healthcheck

from flask import Flask, jsonify
from functools import wraps
import time

app = Flask(__name__)

# 健康检查端点
@app.route('/health')
def health_check():
    """健康检查"""
    checks = {
        'status': 'healthy',
        'timestamp': datetime.now().isoformat(),
        'checks': {}
    }
    
    # 检查数据库
    try:
        db.execute('SELECT 1')
        checks['checks']['database'] = 'ok'
    except Exception as e:
        checks['checks']['database'] = f'error: {e}'
        checks['status'] = 'unhealthy'
    
    # 检查Redis
    try:
        redis.ping()
        checks['checks']['redis'] = 'ok'
    except Exception as e:
        checks['checks']['redis'] = f'error: {e}'
        checks['status'] = 'unhealthy'
    
    status_code = 200 if checks['status'] == 'healthy' else 503
    return jsonify(checks), status_code

# 熔断器实现
class CircuitBreaker:
    """熔断器，防止级联故障"""
    
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = 'closed'  # closed, open, half_open
    
    def call(self, func, *args, **kwargs):
        if self.state == 'open':
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = 'half_open'
            else:
                raise Exception('Circuit breaker is open')
        
        try:
            result = func(*args, **kwargs)
            if self.state == 'half_open':
                self.state = 'closed'
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = 'open'
            
            raise e

# 使用熔断器
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

@app.route('/api/external')
def call_external_api():
    return breaker.call(external_api_call)