Data Map

1
from signalwire_agents.core.data_map import DataMap
2
3
class DataMap:
4
    """Builder class for creating SWAIG data_map configurations."""

1
DataMap(function_name: str)

1
def purpose(self, description: str) -> 'DataMap'
2
def description(self, description: str) -> 'DataMap'  # Alias

1
data_map = DataMap("get_weather").purpose("Get current weather for a city")

1
def parameter(
2
    self,
3
    name: str,                      # Parameter name
4
    param_type: str,                # JSON schema type
5
    description: str,               # Parameter description
6
    required: bool = False,         # Is required
7
    enum: Optional[List[str]] = None  # Allowed values
8
) -> 'DataMap'

1
data_map = (
2
    DataMap("search")
3
    .purpose("Search for items")
4
    .parameter("query", "string", "Search query", required=True)
5
    .parameter("limit", "integer", "Max results", required=False)
6
    .parameter("category", "string", "Category filter",
7
               enum=["electronics", "clothing", "food"])
8
)

1
def webhook(
2
    self,
3
    method: str,                          # HTTP method
4
    url: str,                             # API endpoint
5
    headers: Optional[Dict[str, str]] = None,  # HTTP headers
6
    form_param: Optional[str] = None,     # Form parameter name
7
    input_args_as_params: bool = False,   # Merge args to params
8
    require_args: Optional[List[str]] = None  # Required args
9
) -> 'DataMap'

1
data_map = (
2
    DataMap("get_weather")
3
    .purpose("Get weather information")
4
    .parameter("city", "string", "City name", required=True)
5
    .webhook("GET", "https://api.weather.com/v1/current?q=${enc:args.city}&key=API_KEY")
6
)

1
def body(self, data: Dict[str, Any]) -> 'DataMap'

1
data_map = (
2
    DataMap("create_ticket")
3
    .purpose("Create support ticket")
4
    .parameter("subject", "string", "Ticket subject", required=True)
5
    .parameter("message", "string", "Ticket message", required=True)
6
    .webhook("POST", "https://api.support.com/tickets",
7
             headers={"Authorization": "Bearer TOKEN"})
8
    .body({
9
        "subject": "${args.subject}",
10
        "body": "${args.message}",
11
        "priority": "normal"
12
    })
13
)

1
def params(self, data: Dict[str, Any]) -> 'DataMap'

1
def output(self, result: SwaigFunctionResult) -> 'DataMap'

1
from signalwire_agents.core.function_result import SwaigFunctionResult
2
3
data_map = (
4
    DataMap("get_weather")
5
    .purpose("Get weather")
6
    .parameter("city", "string", "City", required=True)
7
    .webhook("GET", "https://api.weather.com/current?q=${enc:args.city}")
8
    .output(SwaigFunctionResult(
9
        "The weather in ${args.city} is ${response.condition} with a temperature of ${response.temp}°F"
10
    ))
11
)

1
def fallback_output(self, result: SwaigFunctionResult) -> 'DataMap'

1
data_map = (
2
    DataMap("search")
3
    .purpose("Search multiple sources")
4
    .webhook("GET", "https://api.primary.com/search?q=${enc:args.query}")
5
    .output(SwaigFunctionResult("Found: ${response.title}"))
6
    .webhook("GET", "https://api.backup.com/search?q=${enc:args.query}")
7
    .output(SwaigFunctionResult("Backup result: ${response.title}"))
8
    .fallback_output(SwaigFunctionResult("Sorry, search is unavailable"))
9
)

1
def expression(
2
    self,
3
    test_value: str,                        # Template to test
4
    pattern: Union[str, Pattern],           # Regex pattern
5
    output: SwaigFunctionResult,            # Match output
6
    nomatch_output: Optional[SwaigFunctionResult] = None  # No-match output
7
) -> 'DataMap'

1
data_map = (
2
    DataMap("control_playback")
3
    .purpose("Control media playback")
4
    .parameter("command", "string", "Playback command", required=True)
5
    .expression(
6
        "${args.command}",
7
        r"play|start",
8
        SwaigFunctionResult("Starting playback").add_action("playback_bg", "music.mp3")
9
    )
10
    .expression(
11
        "${args.command}",
12
        r"stop|pause",
13
        SwaigFunctionResult("Stopping playback").add_action("stop_playback_bg", True)
14
    )
15
)

1
def foreach(self, foreach_config: Dict[str, Any]) -> 'DataMap'

1
data_map = (
2
    DataMap("search_products")
3
    .purpose("Search product catalog")
4
    .parameter("query", "string", "Search query", required=True)
5
    .webhook("GET", "https://api.store.com/products?q=${enc:args.query}")
6
    .foreach({
7
        "input_key": "products",
8
        "output_key": "product_list",
9
        "max": 3,
10
        "append": "- ${this.name}: $${this.price}\n"
11
    })
12
    .output(SwaigFunctionResult("Found products:\n${product_list}"))
13
)

1
def webhook_expressions(
2
    self,
3
    expressions: List[Dict[str, Any]]
4
) -> 'DataMap'

1
from signalwire_agents import AgentBase
2
from signalwire_agents.core.data_map import DataMap
3
from signalwire_agents.core.function_result import SwaigFunctionResult
4
5
agent = AgentBase(name="weather-agent")
6
7
## Create DataMap
8
weather_map = (
9
    DataMap("get_weather")
10
    .purpose("Get current weather for a location")
11
    .parameter("city", "string", "City name", required=True)
12
    .webhook("GET", "https://api.weather.com/v1/current?q=${enc:args.city}&key=YOUR_KEY")
13
    .output(SwaigFunctionResult(
14
        "The weather in ${args.city} is ${response.current.condition.text} "
15
        "with ${response.current.temp_f}°F"
16
    ))
17
)
18
19
## Register with agent - convert DataMap to SWAIG function dictionary
20
agent.register_swaig_function(weather_map.to_swaig_function())

1
#!/usr/bin/env python3
2
## datamap_api_agent.py - Agent using DataMap for API calls
3
from signalwire_agents import AgentBase
4
from signalwire_agents.core.data_map import DataMap
5
from signalwire_agents.core.function_result import SwaigFunctionResult
6
7
agent = AgentBase(name="api-agent", route="/api")
8
agent.add_language("English", "en-US", "rime.spore")
9
10
## Weather lookup
11
weather = (
12
    DataMap("check_weather")
13
    .purpose("Check weather conditions")
14
    .parameter("location", "string", "City or zip code", required=True)
15
    .webhook("GET", "https://api.weather.com/v1/current?q=${enc:args.location}")
16
    .output(SwaigFunctionResult(
17
        "Current conditions in ${args.location}: ${response.condition}, ${response.temp}°F"
18
    ))
19
    .fallback_output(SwaigFunctionResult("Weather service is currently unavailable"))
20
)
21
22
## Order status lookup
23
order_status = (
24
    DataMap("check_order")
25
    .purpose("Check order status")
26
    .parameter("order_id", "string", "Order number", required=True)
27
    .webhook("GET", "https://api.orders.com/status/${enc:args.order_id}",
28
             headers={"Authorization": "Bearer ${env.API_KEY}"})
29
    .output(SwaigFunctionResult(
30
        "Order ${args.order_id}: ${response.status}. "
31
        "Expected delivery: ${response.delivery_date}"
32
    ))
33
)
34
35
## Expression-based control
36
volume_control = (
37
    DataMap("set_volume")
38
    .purpose("Control audio volume")
39
    .parameter("level", "string", "Volume level", required=True)
40
    .expression("${args.level}", r"high|loud|up",
41
                SwaigFunctionResult("Volume increased").add_action("volume", 100))
42
    .expression("${args.level}", r"low|quiet|down",
43
                SwaigFunctionResult("Volume decreased").add_action("volume", 30))
44
    .expression("${args.level}", r"mute|off",
45
                SwaigFunctionResult("Audio muted").add_action("mute", True))
46
)
47
48
## Register all - convert DataMap to SWAIG function dictionary
49
agent.register_swaig_function(weather.to_swaig_function())
50
agent.register_swaig_function(order_status.to_swaig_function())
51
agent.register_swaig_function(volume_control.to_swaig_function())
52
53
if __name__ == "__main__":
54
    agent.run()