Custom Skills

1
## my_custom_skill/skill.py
2
3
from typing import List, Dict, Any
4
from signalwire_agents.core.skill_base import SkillBase
5
from signalwire_agents.core.function_result import SwaigFunctionResult
6
7
8
class GreetingSkill(SkillBase):
9
    """A skill that provides personalized greetings"""
10
11
    # Required class attributes
12
    SKILL_NAME = "greeting"
13
    SKILL_DESCRIPTION = "Provides personalized greetings"
14
    SKILL_VERSION = "1.0.0"
15
16
    # Optional requirements
17
    REQUIRED_PACKAGES = []
18
    REQUIRED_ENV_VARS = []
19
20
    def setup(self) -> bool:
21
        """Initialize the skill. Return True if successful."""
22
        # Get configuration parameter with default
23
        self.greeting_style = self.params.get("style", "friendly")
24
        return True
25
26
    def register_tools(self) -> None:
27
        """Register SWAIG tools with the agent."""
28
        self.define_tool(
29
            name="greet_user",
30
            description="Generate a personalized greeting",
31
            parameters={
32
                "name": {
33
                    "type": "string",
34
                    "description": "Name of the person to greet"
35
                }
36
            },
37
            handler=self.greet_handler
38
        )
39
40
    def greet_handler(self, args, raw_data):
41
        """Handle greeting requests."""
42
        name = args.get("name", "friend")
43
44
        if self.greeting_style == "formal":
45
            greeting = f"Good day, {name}. How may I assist you?"
46
        else:
47
            greeting = f"Hey {name}! Great to hear from you!"
48
49
        return SwaigFunctionResult(greeting)

1
def setup(self) -> bool:
2
    """
3
    Initialize the skill.
4
5
    Returns:
6
        True if setup successful, False otherwise
7
    """
8
    # Validate packages are installed
9
    if not self.validate_packages():
10
        return False
11
12
    # Validate environment variables
13
    if not self.validate_env_vars():
14
        return False
15
16
    # Initialize from parameters
17
    self.api_url = self.params.get("api_url", "https://api.example.com")
18
    self.timeout = self.params.get("timeout", 30)
19
20
    # Any other initialization
21
    return True

1
def register_tools(self) -> None:
2
    """Register all tools this skill provides."""
3
4
    self.define_tool(
5
        name="my_function",
6
        description="Does something useful",
7
        parameters={
8
            "param1": {
9
                "type": "string",
10
                "description": "First parameter"
11
            },
12
            "param2": {
13
                "type": "integer",
14
                "description": "Second parameter"
15
            }
16
        },
17
        handler=self.my_handler
18
    )
19
20
    # Register multiple tools if needed
21
    self.define_tool(
22
        name="another_function",
23
        description="Does something else",
24
        parameters={},
25
        handler=self.another_handler
26
    )

1
def get_hints(self) -> List[str]:
2
    """Return words to improve speech recognition."""
3
    return ["greeting", "hello", "hi", "welcome"]

1
def get_prompt_sections(self) -> List[Dict[str, Any]]:
2
    """Return prompt sections for the agent."""
3
    return [
4
        {
5
            "title": "Greeting Capability",
6
            "body": "You can greet users by name.",
7
            "bullets": [
8
                "Use greet_user when someone introduces themselves",
9
                "Match the greeting style to the conversation tone"
10
            ]
11
        }
12
    ]

1
def get_global_data(self) -> Dict[str, Any]:
2
    """Return data to add to global context."""
3
    return {
4
        "greeting_skill_enabled": True,
5
        "greeting_style": self.greeting_style
6
    }

1
def cleanup(self) -> None:
2
    """Clean up when skill is removed."""
3
    # Close connections, release resources
4
    if hasattr(self, "connection"):
5
        self.connection.close()

1
@classmethod
2
def get_parameter_schema(cls) -> Dict[str, Dict[str, Any]]:
3
    """Define the parameters this skill accepts."""
4
    # Start with base schema
5
    schema = super().get_parameter_schema()
6
7
    # Add skill-specific parameters
8
    schema.update({
9
        "style": {
10
            "type": "string",
11
            "description": "Greeting style",
12
            "default": "friendly",
13
            "enum": ["friendly", "formal", "casual"],
14
            "required": False
15
        },
16
        "api_key": {
17
            "type": "string",
18
            "description": "API key for external service",
19
            "required": True,
20
            "hidden": True,
21
            "env_var": "MY_SKILL_API_KEY"
22
        }
23
    })
24
25
    return schema

1
class MultiInstanceSkill(SkillBase):
2
    SKILL_NAME = "multi_search"
3
    SKILL_DESCRIPTION = "Searchable with multiple instances"
4
    SKILL_VERSION = "1.0.0"
5
6
    # Enable multiple instances
7
    SUPPORTS_MULTIPLE_INSTANCES = True
8
9
    def get_instance_key(self) -> str:
10
        """Return unique key for this instance."""
11
        tool_name = self.params.get("tool_name", self.SKILL_NAME)
12
        return f"{self.SKILL_NAME}_{tool_name}"
13
14
    def setup(self) -> bool:
15
        self.tool_name = self.params.get("tool_name", "search")
16
        return True
17
18
    def register_tools(self) -> None:
19
        # Use custom tool name
20
        self.define_tool(
21
            name=self.tool_name,
22
            description="Search function",
23
            parameters={
24
                "query": {"type": "string", "description": "Search query"}
25
            },
26
            handler=self.search_handler
27
        )

1
#!/usr/bin/env python3
2
## product_search_skill.py - Custom skill for product search
3
from typing import List, Dict, Any
4
import requests
5
6
from signalwire_agents.core.skill_base import SkillBase
7
from signalwire_agents.core.function_result import SwaigFunctionResult
8
9
10
class ProductSearchSkill(SkillBase):
11
    """Search product catalog"""
12
13
    SKILL_NAME = "product_search"
14
    SKILL_DESCRIPTION = "Search and lookup products in catalog"
15
    SKILL_VERSION = "1.0.0"
16
    REQUIRED_PACKAGES = ["requests"]
17
    REQUIRED_ENV_VARS = []
18
    SUPPORTS_MULTIPLE_INSTANCES = False
19
20
    def setup(self) -> bool:
21
        if not self.validate_packages():
22
            return False
23
24
        self.api_url = self.params.get("api_url")
25
        self.api_key = self.params.get("api_key")
26
27
        if not self.api_url or not self.api_key:
28
            self.logger.error("api_url and api_key are required")
29
            return False
30
31
        return True
32
33
    def register_tools(self) -> None:
34
        self.define_tool(
35
            name="search_products",
36
            description="Search for products by name or category",
37
            parameters={
38
                "query": {
39
                    "type": "string",
40
                    "description": "Search term"
41
                },
42
                "category": {
43
                    "type": "string",
44
                    "description": "Product category filter",
45
                    "enum": ["electronics", "clothing", "home", "all"]
46
                }
47
            },
48
            handler=self.search_handler
49
        )
50
51
        self.define_tool(
52
            name="get_product_details",
53
            description="Get details for a specific product",
54
            parameters={
55
                "product_id": {
56
                    "type": "string",
57
                    "description": "Product ID"
58
                }
59
            },
60
            handler=self.details_handler
61
        )
62
63
    def search_handler(self, args, raw_data):
64
        query = args.get("query", "")
65
        category = args.get("category", "all")
66
67
        try:
68
            response = requests.get(
69
                f"{self.api_url}/search",
70
                params={"q": query, "cat": category},
71
                headers={"Authorization": f"Bearer {self.api_key}"},
72
                timeout=10
73
            )
74
            response.raise_for_status()
75
            data = response.json()
76
77
            products = data.get("products", [])
78
            if not products:
79
                return SwaigFunctionResult(f"No products found for '{query}'")
80
81
            result = f"Found {len(products)} products:\n"
82
            for p in products[:5]:
83
                result += f"- {p['name']} (${p['price']})\n"
84
85
            return SwaigFunctionResult(result)
86
87
        except Exception as e:
88
            self.logger.error(f"Search failed: {e}")
89
            return SwaigFunctionResult("Product search is temporarily unavailable")
90
91
    def details_handler(self, args, raw_data):
92
        product_id = args.get("product_id")
93
94
        try:
95
            response = requests.get(
96
                f"{self.api_url}/products/{product_id}",
97
                headers={"Authorization": f"Bearer {self.api_key}"},
98
                timeout=10
99
            )
100
            response.raise_for_status()
101
            product = response.json()
102
103
            return SwaigFunctionResult(
104
                f"{product['name']}: {product['description']}. "
105
                f"Price: ${product['price']}. In stock: {product['stock']}"
106
            )
107
108
        except Exception as e:
109
            self.logger.error(f"Details lookup failed: {e}")
110
            return SwaigFunctionResult("Could not retrieve product details")
111
112
    def get_hints(self) -> List[str]:
113
        return ["product", "search", "find", "lookup", "catalog"]
114
115
    def get_prompt_sections(self) -> List[Dict[str, Any]]:
116
        return [
117
            {
118
                "title": "Product Search",
119
                "body": "You can search the product catalog.",
120
                "bullets": [
121
                    "Use search_products to find products",
122
                    "Use get_product_details for specific items"
123
                ]
124
            }
125
        ]
126
127
    @classmethod
128
    def get_parameter_schema(cls) -> Dict[str, Dict[str, Any]]:
129
        schema = super().get_parameter_schema()
130
        schema.update({
131
            "api_url": {
132
                "type": "string",
133
                "description": "Product catalog API URL",
134
                "required": True
135
            },
136
            "api_key": {
137
                "type": "string",
138
                "description": "API authentication key",
139
                "required": True,
140
                "hidden": True
141
            }
142
        })
143
        return schema

1
from signalwire_agents.skills.registry import skill_registry
2
3
## Add your skills directory
4
skill_registry.add_skill_directory("/path/to/my_skills")
5
6
## Now use in agent
7
class MyAgent(AgentBase):
8
    def __init__(self):
9
        super().__init__(name="my-agent")
10
        self.add_language("English", "en-US", "rime.spore")
11
12
        self.add_skill("product_search", {
13
            "api_url": "https://api.mystore.com",
14
            "api_key": "secret"
15
        })

1
# test_my_skill.py
2
from my_skills.product_search.skill import ProductSearchSkill
3
4
# Create a mock agent for testing
5
class MockAgent:
6
    def define_tool(self, **kwargs):
7
        print(f"Registered tool: {kwargs['name']}")
8
9
    class log:
10
        @staticmethod
11
        def info(msg): print(f"INFO: {msg}")
12
        @staticmethod
13
        def error(msg): print(f"ERROR: {msg}")
14
15
# Test setup
16
skill = ProductSearchSkill(MockAgent())
17
skill.params = {"api_url": "http://test", "api_key": "test"}
18
assert skill.setup() == True
19
20
# Test tools register
21
skill.register_tools()

1
from signalwire_agents.skills.registry import skill_registry
2
3
# Add and validate your skills
4
skill_registry.add_skill_directory("/path/to/my_skills")
5
6
# Check it loaded
7
available = skill_registry.list_available_skills()
8
print(f"Available skills: {available}")

1
skill_registry.add_skill_directory("/path/to/my_company_skills")

1
# setup.py or pyproject.toml
2
setup(
3
    name="my_company_skills",
4
    entry_points={
5
        "signalwire_agents.skills": [
6
            "product_search = my_company_skills.product_search.skill:ProductSearchSkill",
7
            "crm_integration = my_company_skills.crm_integration.skill:CRMSkill",
8
        ]
9
    }
10
)