Claude Code中Agent Skill的实战入门：从零搭建智能代理系统

1次阅读

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

Agent Skill 是 Claude Code 平台的核心能力单元，理解其设计理念是开发的前提。让我们用快递员来类比：

事件驱动：就像快递员根据订单通知行动，Agent 通过事件触发技能执行（如新订单到达、用户取消请求）
状态管理：快递员需要记住包裹当前在哪个站点，Agent 同样需要维护上下文状态（比如订单处理到哪个环节）
技能编排：复杂的快递配送需要分步骤进行，Agent 也能组合多个子技能完成复杂任务链

安装基础环境（推荐 Python 3.8+）：

pip install claude-sdk==2.3.0  # 注意版本兼容性

权限申请流程：
在 Claude 控制台创建新 Agent
申请 event:order.* 事件订阅权限
获取 API 密钥并配置到环境变量

验证安装：

import claude
print(claude.__version__)  # 应输出 2.3.0

from claude import Agent, Event

class OrderTracker(Agent):
    def __init__(self):
        super().__init__(
            name='order_tracker',
            skills=['order_status_update', 'payment_timeout_check'],
            version='1.0'
        )
        # 状态缓存初始化
        self.order_cache = {}  

    @skill('order_status_update')
    async def handle_order_event(self, event: Event):
        """处理订单状态变更事件"""
        try:
            order_id = event.data['order_id']
            self.order_cache[order_id] = event.data['status']
            self.logger.info(f'Order {order_id} updated to {event.data["status"]}')

            # 触发后续处理（如发货通知）if event.data['status'] == 'paid':
                await self.emit('order.paid', event.data)

        except KeyError as e:
            self.logger.error(f'Invalid event format: {str(e)}')

    @skill('payment_timeout_check')
    async def check_payment_timeout(self):
        """定时检查未支付订单"""
        unpaid_orders = [oid for oid, stat in self.order_cache.items() 
                        if stat == 'unpaid']
        for order_id in unpaid_orders:
            # 模拟重试逻辑
            retry_count = self.order_cache.get(f'{order_id}_retry', 0)
            if retry_count < 3:
                self.order_cache[f'{order_id}_retry'] = retry_count + 1
                await self.notify_user(order_id)

使用 async/await 处理 IO 密集型操作
通过 self.emit() 触发后续事件

错误重试示例：

async def notify_user(self, order_id):
    try:
        # 模拟通知 API 调用
        response = await http.post('/notify', json={
            'order_id': order_id,
            'template': 'payment_reminder'
        })
        if response.status != 200:
            raise Exception('Notification failed')
    except Exception as e:
        self.logger.warning(f'Retrying notification for {order_id}')
        await asyncio.sleep(5)
        await self.notify_user(order_id)  # 递归重试

预热依赖库 ：在__init__ 中提前加载常用模块

保持最小内存：定期清理完成状态的订单数据

def cleanup_cache(self):
    expired = [oid for oid, stat in self.order_cache.items() 
              if stat in ('completed', 'cancelled')]
    for oid in expired:
        del self.order_cache[oid]

设置缓存上限：

MAX_CACHE_SIZE = 1000
if len(self.order_cache) > MAX_CACHE_SIZE:
    self.cleanup_cache()

使用弱引用处理大型附件

检查 @skill 装饰器的参数是否与声明一致
确认 Claude 控制台已启用该技能

确保事件类型订阅正确：

class OrderTracker(Agent):
    subscriptions = ['order.created', 'order.updated']  # 必须包含通配符

添加事件日志追踪：

@middleware
async def log_event(self, event, next_handler):
    self.logger.debug(f'Incoming event: {event.type}')
    return await next_handler(event)

现在你已经掌握了单 Agent 的开发方法，可以思考更复杂的场景：当订单 Agent 需要与库存 Agent、物流 Agent 协作时，如何设计跨 Agent 通信方案？考虑以下维度：