OpenClawでカスタムスキルを自作する実践的ガイド

本記事では、OpenClawの拡張スキル機能を使って業務に特化したカスタムスキルを自作・カスタマイズする手法を解説します。実装例を交えながら、すぐに本番環境で活用できる実践的なステップバイステップアプローチをご紹介します。

OpenClawスキルの基本概念と実装環境

OpenClawはAIエージェント向けのツール統合フレームワークであり、スキルは特定のタスクを実行するための再利用可能なモジュールです。カスタムスキルを構築することで、既存APIやビジネスロジックをエージェントに統合できます。

テスト環境: macOS 14.3 / Python 3.11 / OpenClaw 0.4.2 で動作確認済み

スキルの3つの構成要素

• スキル定義: スキルの名前、説明、入出力スキーマ
• 実装関数: 実際のビジネスロジックを実行するPython関数
• 統合レイヤー: OpenClawエージェントへの登録・呼び出しインターフェース

シンプルなカスタムスキルを5分で実装する

ステップ1: スキルクラスの基本骨組みを作成

最初に、OpenClawのBaseSkillクラスを継承してカスタムスキルを作成します。以下は、顧客データベースから特定の顧客情報を取得するスキルの例です。

from openclaw.skills import BaseSkill
from pydantic import BaseModel, Field
from typing import Optional

# 入力スキーマを定義
class GetCustomerInput(BaseModel):
    customer_id: str = Field(..., description="顧客ID（例: CUST001）")
    include_history: bool = Field(
        default=False, 
        description="購買履歴を含めるかどうか"
    )

# 出力スキーマを定義
class GetCustomerOutput(BaseModel):
    customer_id: str
    name: str
    email: str
    purchase_history: Optional[list] = None
    status: str  # "active", "inactive", "suspended"

# カスタムスキルクラス
class GetCustomerSkill(BaseSkill):
    """顧客情報を取得するカスタムスキル"""
    
    def __init__(self):
        super().__init__(
            name="get_customer",
            description="顧客IDから顧客情報を取得し、オプションで購買履歴も取得する",
            input_schema=GetCustomerInput,
            output_schema=GetCustomerOutput
        )
    
    async def execute(self, input_data: GetCustomerInput) -> GetCustomerOutput:
        """スキルの実行ロジック"""
        # 実装例: データベースから顧客情報を取得
        customer_data = await self._fetch_from_db(input_data.customer_id)
        
        if not customer_data:
            raise ValueError(f"顧客IDが見つかりません: {input_data.customer_id}")
        
        result = GetCustomerOutput(
            customer_id=customer_data["id"],
            name=customer_data["name"],
            email=customer_data["email"],
            status=customer_data["status"]
        )
        
        # 購買履歴の取得がリクエストされている場合
        if input_data.include_history:
            result.purchase_history = await self._fetch_purchase_history(
                input_data.customer_id
            )
        
        return result
    
    async def _fetch_from_db(self, customer_id: str) -> dict:
        """実装例: ここでは静的データを返す"""
        # 実際の環境ではSQLやORM（SQLAlchemy等）を使用
        mock_data = {
            "CUST001": {
                "id": "CUST001",
                "name": "田中太郎",
                "email": "tanaka@example.com",
                "status": "active"
            }
        }
        return mock_data.get(customer_id)
    
    async def _fetch_purchase_history(self, customer_id: str) -> list:
        """購買履歴を取得"""
        return [
            {"order_id": "ORD001", "amount": 50000, "date": "2024-01-15"},
            {"order_id": "ORD002", "amount": 30000, "date": "2024-02-20"}
        ]

ステップ2: エージェントにスキルを登録する

from openclaw.agent import Agent
from openclaw.llm import OpenAILLM

# LLMの初期化
llm = OpenAILLM(model="gpt-4", api_key="your-api-key")

# エージェントの作成
agent = Agent(llm=llm)

# カスタムスキルをインスタンス化して登録
customer_skill = GetCustomerSkill()
agent.register_skill(customer_skill)

# 別のビルトインスキルも登録可能
agent.register_builtin_skill("web_search")
agent.register_builtin_skill("send_email")

ステップ3: スキルの実行とエラーハンドリング

import asyncio

async def main():
    # スキルの実行
    try:
        # ユーザーリクエストからスキルが自動的に選択・実行される
        result = await agent.run(
            "顧客CUST001の情報と購買履歴を取得してください"
        )
        print(f"実行結果: {result}")
    except ValueError as e:
        print(f"エラー: {e}")
    except Exception as e:
        print(f"予期しないエラーが発生しました: {e}")

# 実行
asyncio.run(main())

よくあるハマりポイントと解決策

問題1: スキーマ定義の型ミスマッチ

入力スキーマと実装関数の引数型が一致していないと、ランタイムエラーが発生します。PydanticのBaseModelを必ず使用し、型ヒントを明示的に指定してください。

# ❌ 誤り: 型が曖昧
class BadInput(BaseModel):
    data = Field(...)  # 型が指定されていない

# ✅ 正解: 型を明示的に指定
class GoodInput(BaseModel):
    customer_id: str = Field(..., description="顧客ID")
    amount: float = Field(..., description="金額")

問題2: 非同期処理（async/await）の忘れ

OpenClawのスキルは非同期で実行されることが前提です。execute()メソッドは必ずasync defで定義し、データベースアクセスやAPI呼び出しはawaitを使用してください。

# ❌ 誤り: 同期関数
def execute(self, input_data):
    data = self._fetch_from_db(input_data.id)  # ブロッキング!
    return data

# ✅ 正解: 非同期関数
async def execute(self, input_data):
    data = await self._fetch_from_db(input_data.id)  # ノンブロッキング
    return data

問題3: APIレート制限とタイムアウト

外部API呼び出しを含むスキルの場合、タイムアウトとリトライロジックを実装してください。

import aiohttp
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

class ExternalAPISkill(BaseSkill):
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    async def _call_external_api(self, endpoint: str, timeout: int = 10):
        """リトライ機能付きのAPI呼び出し"""
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(endpoint, timeout=timeout) as response:
                    if response.status == 200:
                        return await response.json()
                    else:
                        raise Exception(f"API Error: {response.status}")
        except asyncio.TimeoutError:
            print("APIタイムアウト - リトライ中...")
            raise

複雑なビジネスロジックを統合する実例

複数ステップを組み合わせたスキル

以下は、注文作成スキルの例です。顧客確認→在庫確認→注文記録という複数ステップを含んでいます。

class CreateOrderSkill(BaseSkill):
    """注文作成スキル（複数ステップ）"""
    
    def __init__(self, customer_skill: GetCustomerSkill, inventory_skill):
        super().__init__(
            name="create_order",
            description="顧客の注文を作成する",
            input_schema=CreateOrderInput,
            output_schema=CreateOrderOutput
        )
        self.customer_skill = customer_skill
        self.inventory_skill = inventory_skill
    
    async def execute(self, input_data: CreateOrderInput) -> CreateOrderOutput:
        # ステップ1: 顧客情報の確認
        customer_result = await self.customer_skill.execute(
            GetCustomerInput(customer_id=input_data.customer_id)
        )
        
        if customer_result.status != "active":
            raise ValueError(f"顧客ステータスが無効です: {customer_result.status}")
        
        # ステップ2: 在庫確認
        inventory_result = await self.inventory_skill.execute(
            CheckInventoryInput(product_id=input_data.product_id)
        )
        
        if inventory_result.available_quantity < input_data.quantity:
            raise ValueError("在庫不足です")
        
        # ステップ3: 注文を記録
        order_id = await self._save_order_to_db(input_data)
        
        return CreateOrderOutput(
            order_id=order_id,
            customer_name=customer_result.name,
            product_id=input_data.product_id,
            quantity=input_data.quantity,
            status="confirmed"
        )

スキルのテストと品質保証

ユニットテストの実装例

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_get_customer_skill():
    """顧客情報取得スキルのテスト"""
    skill = GetCustomerSkill()
    
    # モックデータを設定
    with patch.object(skill, '_fetch_from_db', new_callable=AsyncMock) as mock_db:
        mock_db.return_value = {
            "id": "CUST001",
            "name": "テスト太郎",
            "email": "test@example.com",
            "status": "active"
        }
        
        # スキルを実行
        result = await skill.execute(
            GetCustomerInput(customer_id="CUST001", include_history=False)
        )
        
        # 結果をアサート
        assert result.customer_id == "CUST001"
        assert result.name == "テスト太郎"
        assert result.purchase_history is None

@pytest.mark.asyncio
async def test_get_customer_not_found():
    """顧客が見つからない場合のテスト"""
    skill = GetCustomerSkill()
    
    with patch.object(skill, '_fetch_from_db', new_callable=AsyncMock) as mock_db:
        mock_db.return_value = None
        
        with pytest.raises(ValueError, match="顧客IDが見つかりません"):
            await skill.execute(GetCustomerInput(customer_id="INVALID"))

本番環境での運用ポイント

ロギングとモニタリング

スキルの実行をトラッキングするため、詳細なロギングを追加してください。

import logging
from datetime import datetime

logger = logging.getLogger(__name__)

class GetCustomerSkill(BaseSkill):
    async def execute(self, input_data: GetCustomerInput) -> GetCustomerOutput:
        start_time = datetime.now()
        logger.info(f"スキル開始: {self.name}, 顧客ID: {input_data.customer_id}")
        
        try:
            result = await self._execute_logic(input_data)
            duration = (datetime.now() - start_time).total_seconds()
            logger.info(f"スキル完了: {self.name}, 実行時間: {duration:.2f}秒")
            return result
        except Exception as e:
            logger.error(f"スキルエラー: {self.name}, エラー: {str(e)}")
            raise
    
    async def _execute_logic(self, input_data):
        # 実装

OpenClawスキルと他の選択肢の比較