DeepAgents Skill调用API入门指南：从零搭建到生产环境部署

1次阅读

共计 3943 个字符，预计需要花费 10 分钟才能阅读完成。

DeepAgents Skill 调用 API（以下简称 Skill API）适用于需要灵活接入 AI 能力的场景。比如：

智能客服系统：当用户咨询商品售后政策时，系统通过 Skill API 调用「政策查询」技能，实时返回结构化结果（如退货期限、保修范围）
自动化流程：电商订单系统中，自动调用「地址校验」技能核验收件人信息，避免因地址错误导致的物流失败

维度	REST	gRPC
协议	HTTP/1.1	HTTP/2
数据格式	JSON	Protocol Buffers
适用场景	简单技能调用	高频 / 低延迟技能组合调用
开发效率	调试方便（可直接用 curl 测试）	需要编译 proto 文件
性能基准	100ms 延迟（测试环境：2 核 4G）	35ms 延迟（同等测试环境）

推荐选择原则：
– 新手建议从 REST 开始
– 高并发场景优先考虑 gRPC

所有 API 请求需携带身份令牌。以下是 Python 生成 JWT 的示例：

import jwt
import time

# 从控制台获取的密钥
API_KEY = "your_api_key_here"
SECRET = "your_secret_here"

def generate_token():
    """
    生成 JWT 认证令牌
    有效时间：1 小时
    算法：HS256
    """payload = {"api_key": API_KEY,"exp": int(time.time()) + 3600  # 过期时间戳
    }
    return jwt.encode(payload, SECRET, algorithm="HS256")

# 使用示例
token = generate_token()
print(f"Authorization: Bearer {token}")

典型技能调用包含三个要素：

import requests

# 1. 准备请求头
headers = {"Authorization": f"Bearer {token}",
    "Content-Type": "application/json"
}

# 2. 构造请求体
payload = {
    "skill_id": "address_validation",  # 技能 ID
    "params": {"address": "123 Main St, San Francisco"},
    "request_id": "req_123"  # 用于追踪
}

# 3. 发送请求
response = requests.post(
    "https://api.deepagents.com/v1/skills/execute",
    json=payload,
    headers=headers
)

# 4. 处理响应
if response.status_code == 200:
    result = response.json()
    print(f"技能执行成功: {result['data']}")
else:
    print(f"错误码 {response.status_code}: {response.text}")

对于耗时技能（如文档审核），建议使用异步模式：

# 初始请求（添加 async 字段）async_payload = {
    "skill_id": "doc_review",
    "async": True,
    "callback_url": "https://your-server.com/callback"  # 结果回调地址
}

# 轮询检查结果（示例）def check_result(task_id):
    while True:
        resp = requests.get(f"https://api.deepagents.com/v1/tasks/{task_id}", 
                          headers=headers)
        data = resp.json()

        if data["status"] == "completed":
            return data["result"]
        elif data["status"] == "failed":
            raise Exception(data["error"])

        time.sleep(2)  # 每 2 秒检查一次

建议采用指数退避重试：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),  # 最多重试 3 次
    wait=wait_exponential(multiplier=1, max=10)  # 1s, 2s, 4s 间隔
)
def call_skill_with_retry(payload):
    response = requests.post(API_ENDPOINT, json=payload, headers=headers)
    if response.status_code >= 500:
        raise Exception("服务器错误，触发重试")
    return response

使用 circuitbreaker 避免雪崩：

from circuitbreaker import circuit

@circuit(
    failure_threshold=5,  # 连续 5 次失败触发熔断
    recovery_timeout=30   # 30 秒后尝试恢复
)
def protected_skill_call(payload):
    return call_skill_with_retry(payload)

在日志记录前处理敏感字段：

def sanitize_data(data):
    sensitive_fields = ["phone", "id_number"]
    for field in sensitive_fields:
        if field in data:
            data[field] = "***MASKED***"
    return data

# 使用示例
log_data = sanitize_data(raw_response)
logger.info(f"API 响应: {log_data}")

# 带错误处理的请求示例
curl -X POST \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"skill_id":"weather_query","params":{"city":"Beijing"}}' \
  https://api.deepagents.com/v1/skills/execute \
  || echo "API 调用失败，状态码: $?"

import requests
import json

class DeepAgentsClient:
    def __init__(self, api_key, secret):
        self.api_key = api_key
        self.secret = secret
        self.base_url = "https://api.deepagents.com/v1"

    def _get_token(self):
        """私有方法：生成 JWT"""
        payload = {"api_key": self.api_key, "exp": int(time.time()) + 3600}
        return jwt.encode(payload, self.secret, algorithm="HS256")

    def execute_skill(self, skill_id, params, async_mode=False):
        """
        执行技能调用
        :param skill_id: 技能标识符
        :param params: 输入参数字典
        :param async_mode: 是否异步执行
        :return: 技能执行结果
        """url = f"{self.base_url}/skills/execute"headers = {"Authorization": f"Bearer {self._get_token()}","Content-Type":"application/json"
        }

        try:
            payload = {
                "skill_id": skill_id,
                "params": params,
                "async": async_mode
            }

            response = requests.post(url, json=payload, headers=headers)
            response.raise_for_status()  # 自动处理 4xx/5xx 错误
            return response.json()

        except requests.exceptions.RequestException as e:
            print(f"请求失败: {str(e)}")
            return None

# 使用示例
client = DeepAgentsClient("your_key", "your_secret")
result = client.execute_skill("sentiment_analysis", {"text": "这个产品非常好用！"})
print(json.dumps(result, indent=2))