triggers: provider: - provide.*service - publish.*service - expose.*service - create.*service - make.*service - implement.*service consumer: - what services - list services - call.*service - use.*service - query.*data - fetch.*data

Tool Service Hub Skill#

Overview#

Enables subagents to:

1. Provider Mode: Publish local data/capabilities as services for other subagents to call
2. Consumer Mode: Discover and call services on the Hub

---

1. Publishing Services as a Provider#

Complete Code Template#

python

import asyncio
import os
import sys
from pathlib import Path

# === 1. Setup path ===
WORKSPACE_DIR = os.getenv('WORKSPACE_DIR', '/home/t/.openclaw/workspace-subagentX')
sys.path.insert(0, WORKSPACE_DIR)

from client.client import LocalServiceRunner

# === 2. Define your service capability ===

async def your_method(**params):
    """
    Service method
    params: Parameters passed by the caller
    Must return a dict
    """
    # Your business logic here
    result = {"status": "ok", "data": "..."}
    return result

# === 3. Start the service ===

async def main():
    runner = LocalServiceRunner(
        name="your-service-name",      # Service name (English, no spaces)
        description="Service description",  # English description
        hub_url=os.getenv("HUB_WS_URL", "ws://localhost:8765")
    )

    # Register methods (can register multiple)
    runner.register_handler("your_method", your_method)

    print(f"🚀 Starting service...")
    await runner.run()

if __name__ == "__main__":
    asyncio.run(main())

---

2. Calling Services as a Consumer#

Complete Code Template#

python

import asyncio
import os
import sys

WORKSPACE_DIR = os.getenv('WORKSPACE_DIR', '/home/t/.openclaw/workspace-subagentX')
sys.path.insert(0, WORKSPACE_DIR)

from client.skill_client import SkillQueryClient

async def main():
    # 1. Connect to Hub
    client = SkillQueryClient(
        hub_url=os.getenv("HUB_WS_URL", "ws://localhost:8765")
    )
    await client.connect()

    # 2. Discover services
    services = await client.discover()
    print(f"Discovered {len(services)} services")

    # 3. Find target service (filter by name)
    target = None
    target_name = "weather-service"  # Replace with your target service name
    for s in services:
        if target_name in s.get("name", ""):
            target = s
            break

    if not target:
        print(f"Service not found: {target_name}")
        return

    skill_id = target.get("skill_id")
    print(f"Using service: {target.get('name')}, skill_id: {skill_id}")

    # 4. Call the service
    result = await client.call_service(
        service_id=skill_id,
        method="your_method",      # Method name
        params={"key": "value"}    # Parameters
    )

    print(f"Result: {result}")

    await client.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

---

3. Common Data Source Examples#

3.1 File Data Service#

python

from pathlib import Path

DATA_DIR = Path("/path/to/data")  # Change to actual directory

async def list_files(**params):
    ext = params.get("extension", "")
    pattern = f"*{ext}" if ext else "*"
    files = [f.name for f in DATA_DIR.glob(pattern) if f.is_file()]
    return {"files": files[:50], "total": len(files)}

async def read_file(**params):
    filename = params.get("filename")
    if not filename:
        return {"error": "filename is required"}

    filepath = DATA_DIR / filename
    if not filepath.exists():
        return {"error": f"File not found: {filename}"}

    # Read text files directly
    if filepath.suffix == '.txt':
        return {"content": filepath.read_text()[:1000]}

    # Return info for other files
    return {"filename": filename, "size": filepath.stat().st_size}

3.2 API Data Service#

python

import aiohttp

async def fetch_data(**params):
    url = params.get("url")
    if not url:
        return {"error": "url is required"}

    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, timeout=aiohttp.ClientTimeout(10)) as resp:
                data = await resp.json()
        return {"status": resp.status, "data": data}
    except Exception as e:
        return {"error": str(e)}

3.3 Weather Service (wttr.in)#

python

import aiohttp

async def get_weather(**params):
    city = params.get("city", "Shanghai")
    url = f"https://wttr.in/{city}?format=j1"

    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, timeout=aiohttp.ClientTimeout(10)) as resp:
                data = await resp.json()

        # Note: wttr.in returns structure at data.current_condition
        c = data.get("data", {}).get("current_condition", [{}])[0]
        return {
            "city": city,
            "temp": int(c.get("temp_C") or 0),
            "condition": c.get("weatherDesc", [{}])[0].get("value", "Unknown"),
            "humidity": c.get("humidity")
        }
    except Exception as e:
        return {"error": str(e)}

---

4. Workflow Examples#

Combining Multiple Services#

python

async def workflow():
    """Complete workflow combining multiple services"""
    client = SkillQueryClient("ws://localhost:8765")
    await client.connect()

    services = await client.discover()

    # Find required services
    weather = next((s for s in services if "weather" in s.get("name", "")), None)
    images = next((s for s in services if "image" in s.get("name", "")), None)

    results = {}

    # Call weather service
    if weather:
        w = await client.call_service(weather.get("skill_id"), "get_weather", {"city": "Shanghai"})
        results["weather"] = w.get("result", {})

    # Call image service
    if images:
        i = await client.call_service(images.get("skill_id"), "list_images", {"limit": 10})
        results["images"] = i.get("result", {})

    await client.disconnect()
    return results

---

5. Environment Configuration#

Install Dependencies#

bash

pip install websockets aiohttp

Environment Variables#

VariableDefaultDescriptionHUB_WS_URLws://localhost:8765Hub WebSocket addressWORKSPACE_DIR/home/t/.../workspace-subagentXWorking directory

Start Hub Server (Optional)#

bash

cd Claw-Service-Hub
python -m server.main

# WebSocket: ws://0.0.0.0:8765
# REST API: http://0.0.0.0:3765

---

6. Troubleshooting#

Issue 1: ImportError#

Error: ModuleNotFoundError: No module named 'client'

Fix: Set sys.path correctly

python

import os
import sys
WORKSPACE_DIR = os.getenv('WORKSPACE_DIR', '/home/t/.openclaw/workspace-subagentX')
sys.path.insert(0, WORKSPACE_DIR)
from client.client import LocalServiceRunner

Issue 2: Service Registered but Cannot Be Called#

Check:

1. Is the Provider process still running?
2. Is the method name correct (case-sensitive)?
3. Is the parameter format correct?

Issue 3: API Returns None#

Common Cause: Data structure parsing error

Fix: Print raw data first to confirm structure

python

async def get_data(**params):
    url = params.get("url")
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            data = await resp.json()
    print(f"Raw data: {data}")  # Add this debug line
    # Then parse based on actual structure
    return {"data": data}

Issue 4: Service Not Found#

Fix: List all services first

python

services = await client.discover()
for s in services:
    print(f"{s.get('name')}: {s.get('skill_id')}")

Issue 5: Return Value Must Be Dict#

Error: TypeError: ... got an unexpected keyword argument

Fix: Handlers must return dict

python

async def wrong():  # Wrong
    return "string"

async def right(**params):  # Correct
    return {"result": "value"}

Issue 6: Return Value Wrapped in 'result'#

现象: After calling service, returns {'result': {'actual_data': '...'}}

说明: Hub wraps the handler's returned dict in the 'result' field

Fix: Use the return value directly, or extract as needed

python

result = await client.call_service(service_id, "method", params)
# result = {'result': {'temp': 25, 'city': 'Beijing'}}

# Method 1: Use directly (recommended)
data = result  # Already unpacked data

# Method 2: If explicit extraction needed
if 'result' in result:
    data = result['result']

---

7. Minimal Examples#

Provider (5 lines)#

python

import asyncio, os, sys
sys.path.insert(0, os.getenv('WORKSPACE_DIR','.'))
from client.client import LocalServiceRunner

async def hello(**p): return {"msg":"Hello!"}
r = LocalServiceRunner("demo","Demo Service",os.getenv("HUB_WS_URL","ws://localhost:8765"))
r.register_handler("hello", hello)
asyncio.run(r.run())

Consumer (6 lines)#

python

import asyncio, os, sys
sys.path.insert(0, os.getenv('WORKSPACE_DIR','.'))
from client.skill_client import SkillQueryClient

async def main():
    c = SkillQueryClient()
    await c.connect()
    print([s.get("name") for s in await c.discover()])
    await c.disconnect()
asyncio.run(main())

---

8. File Structure#

text

Claw-Service-Hub/
├── client/
│   ├── client.py           # LocalServiceRunner, ToolServiceClient
│   ├── skill_client.py     # SkillQueryClient
│   └── management_client.py
├── skills/
│   └── hub-client/
│       └── SKILL.md        # This file
└── server/
    └── main.py             # Hub Server

---

9. Publishing Flow#

1. Ensure Hub Server is running ( ws://localhost:8765)
2. Provider: Run python your_service.py to register service
3. Consumer: Connect to Hub to discover services
4. Consumer: Call service to get results

---

10. Key Authorization Mechanism (Optional)#

10.1 Overview#

Optional key authorization mechanism for controlling service access.

Features:

• Time dimension: Key validity period (customizable by Provider)
• Count dimension: Maximum call count (customizable by Provider)
• Dual verification: Provider self-management + Hub verification

10.2 Provider Side - Set Lifecycle Policy#

python

from client.client import LocalServiceRunner

# Create service
runner = LocalServiceRunner(
    name="my-protected-service",
    description="Service requiring authorization",
    hub_url=os.getenv("HUB_WS_URL", "ws://localhost:8765")
)

# Set default lifecycle policy
runner.set_lifecycle_policy(
    duration_seconds=3600,  # Default 1 hour validity
    max_calls=100           # Default 100 calls
)

# Set custom policy (optional)
runner.set_custom_policy(
    condition="premium",     # Policy name
    duration_seconds=86400,  # 24 hours
    max_calls=1000           # 1000 calls
)

# Register method
async def get_data(**params):
    return {"data": "secret data"}

runner.register_handler("get_data", get_data)

# Start service
print(f"🚀 Starting authorized service...")
await runner.run()

10.3 Consumer Side - Request Key and Call#

python

from client.skill_client import SkillQueryClient

async def main():
    client = SkillQueryClient(
        hub_url=os.getenv("HUB_WS_URL", "ws://localhost:8765")
    )
    await client.connect()

    # Discover services
    services = await client.discover()
    target = next((s for s in services if "protected" in s.get("name", "")), None)

    if not target:
        print("Service not found")
        return

    service_id = target.get("skill_id")

    # Method 1: Direct call (if service doesn't require Key)
    # result = await client.call_service(service_id, "get_data", {})

    # Method 2: Request Key first, then call (recommended)
    key_info = await client.request_key(
        service_id=service_id,
        purpose="Daily data query"
    )

    if key_info.get("success"):
        key = key_info["key"]
        lifecycle = key_info["lifecycle"]

        print(f"✅ Key obtained successfully")
        print(f"   Key: {key[:20]}...")
        print(f"   Validity: {lifecycle.get('remaining_time')} seconds")
        print(f"   Remaining calls: {lifecycle.get('remaining_calls')} times")

        # Call service with Key
        result = await client.call_service(
            service_id=service_id,
            method="get_data",
            params={},
            key=key  # Carry the Key
        )

        print(f"📥 Result: {result}")
    else:
        print(f"❌ Key request failed: {key_info.get('reason')}")

    await client.disconnect()

asyncio.run(main())

10.4 Handling Key Verification Failure#

python

async def call_with_fallback(client, service_id, method, params):
    """Call with automatic retry"""

    # Try without Key first
    result = await client.call_service(service_id, method, params)

    # Check if Key is required
    if result.get("error") and "Key" in result.get("error", ""):
        print("Key required, requesting authorization...")

        key_info = await client.request_key(service_id, "Auto request")
        if key_info.get("success"):
            key = key_info["key"]
            # Retry with Key
            result = await client.call_service(
                service_id, method, params, key=key
            )

    return result

10.5 Message Protocol#

Message TypeDirectionDescriptionlifecycle_policyProvider→HubRegister lifecycle policykey_requestConsumer→Hub→ProviderRequest Keykey_responseProvider→Hub→ConsumerReturn Key (approve/reject)key_revokeProvider→HubRevoke Keycall_service (with key)Consumer→HubCall service with Key

10.6 Lifecycle Parameters#

python

# Provider registers policy
{
    "duration_seconds": 3600,    # Validity duration (seconds)
    "max_calls": 100,             # Maximum call count
    "custom_policies": {         # Optional: custom policies
        "premium": {
            "duration_seconds": 86400,
            "max_calls": 1000
        }
    }
}

# Key verification result
{
    "valid": True,
    "key": "key_abc123...",
    "lifecycle": {
        "expires_at": "2026-03-20T03:47:00Z",
        "max_calls": 100,
        "call_count": 5,
        "remaining_calls": 95,
        "remaining_time": 3200
    }
}

10.7 Subagent Usage Tips#

For Provider (Service Publisher):

1. If service requires authorization, set reasonable limits in set_lifecycle_policy
2. Recommended defaults: max_calls=100, duration_seconds=3600
3. Can set different policies for different consumers

For Consumer (Service Caller):

1. Try direct call first, request Key if it fails
2. Save Key to avoid repeated requests
3. Check lifecycle.get('remaining_calls') to avoid running out of calls
4. Watch lifecycle.get('remaining_time') and renew as needed

---

License#

MIT License

Copyright (c) 2026 OpenClaw

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.