Skip to main content

Data Map

DataMap API

Summary: API reference for DataMap, enabling serverless REST API integration without webhooks.

Class Definition

from signalwire_agents.core.data_map import DataMap

class DataMap:
"""Builder class for creating SWAIG data_map configurations."""

Overview

DataMap enables SWAIG functions that execute on SignalWire servers without requiring your own webhook endpoints.

Use Cases:

  • Call external APIs directly from SWML
  • Pattern-based responses without API calls
  • Reduce infrastructure requirements
  • Serverless function execution

Constructor

DataMap(function_name: str)

Create a new DataMap builder.

Core Methods

purpose / description

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

Set the function description shown to the AI.

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

parameter

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

Add a function parameter.

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

Parameter Types

TypeJSON SchemaDescription
stringstringText values
integerintegerWhole numbers
numbernumberDecimal numbers
booleanbooleanTrue/False
arrayarrayList of items
objectobjectKey-value pairs

Webhook Methods

webhook

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

Add an API call.

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

body

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

Set request body for POST/PUT.

data_map = (
DataMap("create_ticket")
.purpose("Create support ticket")
.parameter("subject", "string", "Ticket subject", required=True)
.parameter("message", "string", "Ticket message", required=True)
.webhook("POST", "https://api.support.com/tickets",
headers={"Authorization": "Bearer TOKEN"})
.body({
"subject": "${args.subject}",
"body": "${args.message}",
"priority": "normal"
})
)

params

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

Set request parameters (alias for body).

Output Methods

output

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

Set the output for the most recent webhook.

from signalwire_agents.core.function_result import SwaigFunctionResult

data_map = (
DataMap("get_weather")
.purpose("Get weather")
.parameter("city", "string", "City", required=True)
.webhook("GET", "https://api.weather.com/current?q=${enc:args.city}")
.output(SwaigFunctionResult(
"The weather in ${args.city} is ${response.condition} with a temperature of ${response.temp}°F"
))
)

fallback_output

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

Set output when all webhooks fail.

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

Variable Patterns

PatternDescription
${args.param}Function argument value
${enc:args.param}URL-encoded argument (use in webhook URLs)
${lc:args.param}Lowercase argument value
${fmt_ph:args.phone}Format as phone number
${response.field}API response field
${response.arr[0]}Array element in response
${global_data.key}Global session data
${meta_data.key}Call metadata
${this.field}Current item in foreach

Chained Modifiers

Modifiers are applied right-to-left:

PatternResult
${enc:lc:args.param}First lowercase, then URL encode
${lc:enc:args.param}First URL encode, then lowercase

Examples

PatternResult
${args.city}"Seattle" (in body/output)
${enc:args.city}"Seattle" URL-encoded (in URLs)
${lc:args.city}"seattle" (lowercase)
${enc:lc:args.city}"seattle" lowercased then URL-encoded
${fmt_ph:args.phone}"+1 (555) 123-4567"
${response.temp}"65"
${response.items[0].name}"First item"
${global_data.user_id}"user123"

Expression Methods

expression

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

Add pattern-based response (no API call needed).

data_map = (
DataMap("control_playback")
.purpose("Control media playback")
.parameter("command", "string", "Playback command", required=True)
.expression(
"${args.command}",
r"play|start",
SwaigFunctionResult("Starting playback").add_action("playback_bg", "music.mp3")
)
.expression(
"${args.command}",
r"stop|pause",
SwaigFunctionResult("Stopping playback").add_action("stop_playback_bg", True)
)
)

Array Processing

foreach

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

Process array from API response.

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

Foreach Configuration

KeyTypeDescription
input_keystringKey in response containing array
output_keystringVariable name for built string
maxintegerMaximum items to process (optional)
appendstringTemplate for each item

Webhook Expressions

webhook_expressions

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

Add expressions to run after webhook completes.

Registering with Agent

from signalwire_agents import AgentBase
from signalwire_agents.core.data_map import DataMap
from signalwire_agents.core.function_result import SwaigFunctionResult

agent = AgentBase(name="weather-agent")

## Create DataMap
weather_map = (
DataMap("get_weather")
.purpose("Get current weather for a location")
.parameter("city", "string", "City name", required=True)
.webhook("GET", "https://api.weather.com/v1/current?q=${enc:args.city}&key=YOUR_KEY")
.output(SwaigFunctionResult(
"The weather in ${args.city} is ${response.current.condition.text} "
"with ${response.current.temp_f}°F"
))
)

## Register with agent - convert DataMap to SWAIG function dictionary
agent.register_swaig_function(weather_map.to_swaig_function())

Complete Example

#!/usr/bin/env python3
## datamap_api_agent.py - Agent using DataMap for API calls
from signalwire_agents import AgentBase
from signalwire_agents.core.data_map import DataMap
from signalwire_agents.core.function_result import SwaigFunctionResult

agent = AgentBase(name="api-agent", route="/api")
agent.add_language("English", "en-US", "rime.spore")

## Weather lookup
weather = (
DataMap("check_weather")
.purpose("Check weather conditions")
.parameter("location", "string", "City or zip code", required=True)
.webhook("GET", "https://api.weather.com/v1/current?q=${enc:args.location}")
.output(SwaigFunctionResult(
"Current conditions in ${args.location}: ${response.condition}, ${response.temp}°F"
))
.fallback_output(SwaigFunctionResult("Weather service is currently unavailable"))
)

## Order status lookup
order_status = (
DataMap("check_order")
.purpose("Check order status")
.parameter("order_id", "string", "Order number", required=True)
.webhook("GET", "https://api.orders.com/status/${enc:args.order_id}",
headers={"Authorization": "Bearer ${env.API_KEY}"})
.output(SwaigFunctionResult(
"Order ${args.order_id}: ${response.status}. "
"Expected delivery: ${response.delivery_date}"
))
)

## Expression-based control
volume_control = (
DataMap("set_volume")
.purpose("Control audio volume")
.parameter("level", "string", "Volume level", required=True)
.expression("${args.level}", r"high|loud|up",
SwaigFunctionResult("Volume increased").add_action("volume", 100))
.expression("${args.level}", r"low|quiet|down",
SwaigFunctionResult("Volume decreased").add_action("volume", 30))
.expression("${args.level}", r"mute|off",
SwaigFunctionResult("Audio muted").add_action("mute", True))
)

## Register all - convert DataMap to SWAIG function dictionary
agent.register_swaig_function(weather.to_swaig_function())
agent.register_swaig_function(order_status.to_swaig_function())
agent.register_swaig_function(volume_control.to_swaig_function())

if __name__ == "__main__":
agent.run()

When to Use DataMap

ScenarioUse DataMap?Alternative
Simple REST API callsYes-
Pattern-based responsesYes-
Complex business logicNoSWAIG function with webhook
Database accessNoSWAIG function
Multiple conditional pathsMaybeConsider SWAIG for complex logic

See Also

TopicReference
DataMap guideDataMap Functions
SWAIG functionsSWAIG Function API
Function resultsSwaigFunctionResult API
Testing DataMapswaig-test CLI - DataMap functions make live HTTP calls

Troubleshooting

IssueSolution
Variable not substitutingCheck parameter name matches ${args.param} exactly
API returns errorUse fallback_output() to handle failures gracefully
URL encoding issuesUse ${enc:args.param} for URL parameters
Response field not foundCheck API response structure; use ${response.nested.field} for nested data