api_reference.md

# SignalWire SDK - Complete API Reference

This document provides a comprehensive reference for all public APIs in the SignalWire SDK.

<!-- snippet-setup: shared imports the reference examples assume -->
```python
from signalwire import AgentBase, DataMap, FunctionResult, SwaigFunctionResult, SWMLService
from signalwire.core.data_map import create_expression_tool
```

## Table of Contents

1. [AgentBase Class](#agentbase-class) - Core agent functionality
2. [FunctionResult Class](#swaigfunctionresult-class) - SWAIG (SignalWire AI Gateway) function response handling
3. [DataMap Class](#datamap-class) - Serverless API tools that execute on SignalWire's servers
4. [Context System](#context-system) - Structured workflows
5. [State Management](#state-management) - Persistent state
6. [Skills System](#skills-system) - Modular capabilities
7. [Utility Classes](#utility-classes) - Supporting classes

---

## AgentBase Class

The `AgentBase` class is the foundation for creating AI agents. It extends `SWMLService` (the base class for generating SWML documents, where SWML is SignalWire Markup Language) and provides comprehensive functionality for building conversational AI agents.

### Constructor

This is the full constructor signature:

<!-- snippet: no-compile signature-illustration -->
```python
AgentBase(
    name: str,
    route: str = "/",
    host: str = "0.0.0.0",
    port: Optional[int] = None,
    basic_auth: Optional[Tuple[str, str]] = None,
    use_pom: bool = True,
    token_expiry_secs: int = 3600,
    auto_answer: bool = True,
    record_call: bool = False,
    record_format: str = "mp4",
    record_stereo: bool = True,
    default_webhook_url: Optional[str] = None,
    agent_id: Optional[str] = None,
    native_functions: Optional[List[str]] = None,
    schema_path: Optional[str] = None,
    suppress_logs: bool = False,
    enable_post_prompt_override: bool = False,
    check_for_input_override: bool = False,
    config_file: Optional[str] = None,
    schema_validation: bool = True,
    signing_key: Optional[str] = None,
    trust_proxy_for_signature: bool = False,
    swaig_secret: Optional[str] = None
)
```

**Parameters:**
- `name` (str): Human-readable name for the agent
- `route` (str): HTTP route path for the agent (default: "/")
- `host` (str): Host address to bind to (default: "0.0.0.0")
- `port` (Optional[int]): Port number to listen on. If unset, defaults to 3000 at serve/run time.
- `basic_auth` (Optional[Tuple[str, str]]): Username/password for HTTP basic auth
- `use_pom` (bool): Whether to use Prompt Object Model (default: True)
- `token_expiry_secs` (int): Security token expiration time (default: 3600)
- `auto_answer` (bool): Automatically answer incoming calls (default: True)
- `record_call` (bool): Record calls by default (default: False)
- `record_format` (str): Recording format: "mp4", "wav", "mp3" (default: "mp4")
- `record_stereo` (bool): Record in stereo (default: True)
- `default_webhook_url` (Optional[str]): Default webhook URL for functions
- `agent_id` (Optional[str]): Unique identifier for the agent
- `native_functions` (Optional[List[str]]): List of native function names to enable
- `schema_path` (Optional[str]): Path to custom SWML schema file
- `suppress_logs` (bool): Silences a few legacy debug and error log lines, not the SDK's logging generally (default: False). See [Suppressing Logs](swml_service_guide.md#suppressing-logs) for how to control log output.
- `enable_post_prompt_override` (bool): Allow post-prompt URL override (default: False)
- `check_for_input_override` (bool): Allow check-for-input URL override (default: False)
- `config_file` (Optional[str]): Path to JSON configuration file with environment variable substitution support. See [Configuration Guide](configuration.md) for details.
- `schema_validation` (bool): Enable SWML schema validation (default: True). Can also be disabled with the `SWML_SKIP_SCHEMA_VALIDATION=1` environment variable.
- `signing_key` (Optional[str]): SignalWire Signing Key from the Dashboard. When set, the agent requires a valid signature on every POST. Falls back to the `SIGNALWIRE_SIGNING_KEY` environment variable. See the [Security Configuration Guide](security.md#webhook-signature-validation).
- `trust_proxy_for_signature` (bool): Honor `X-Forwarded-Proto`/`X-Forwarded-Host` when reconstructing the URL for signature validation (default: False). Enable only when you control the proxy chain.
- `swaig_secret` (Optional[str]): Secret used to sign this agent's per-call SWAIG function tokens. Falls back to the `SIGNALWIRE_SWAIG_SECRET` environment variable, then a random secret generated per process. See the [Security Configuration Guide](security.md#swaig-function-token-signing).

### Core Methods

#### Deployment and Execution

##### `run(event=None, context=None, force_mode=None, host=None, port=None)`
Auto-detects deployment environment and runs the agent appropriately.

**Parameters:**
- `event`: Event object for serverless environments
- `context`: Context object for serverless environments  
- `force_mode` (str): Force specific mode: "server", "lambda", "cgi", "cloud_function"
- `host` (Optional[str]): Override host address
- `port` (Optional[int]): Override port number

**Usage:**
```python
# Auto-detect environment
agent.run()

# Force server mode
agent.run(force_mode="server", host="localhost", port=8080)

# Lambda handler
def lambda_handler(event, context):
    return agent.run(event, context)
```

##### `serve(host=None, port=None)`
Explicitly run as HTTP server using FastAPI/Uvicorn.

**Parameters:**
- `host` (Optional[str]): Host address to bind to
- `port` (Optional[int]): Port number to listen on

**Usage:**
```python
agent.serve()  # Use constructor defaults
agent.serve(host="0.0.0.0", port=3000)
```

### Prompt Configuration

#### Text-Based Prompts

##### `set_prompt_text(text: str) -> AgentBase`
Set the agent's prompt as raw text.

**Parameters:**
- `text` (str): The complete prompt text

**Usage:**
```python
agent.set_prompt_text("You are a helpful customer service agent.")
```

##### `set_post_prompt(text: str) -> AgentBase`
Set additional text to append after the main prompt.

**Parameters:**
- `text` (str): Text to append after main prompt

**Usage:**
```python
agent.set_post_prompt("Always be polite and professional.")
```

#### LLM Parameter Configuration

##### `set_prompt_llm_params`

Set Language Model parameters for the main prompt. Accepts any parameters which will be passed through to the SignalWire server. The server validates and applies parameters based on the target model's capabilities.

<!-- snippet: no-compile signature-illustration -->
```python
def set_prompt_llm_params(**params) -> AgentBase
```

**Common Parameters:**
- `temperature`: Controls randomness. Lower = more focused
- `top_p`: Nucleus sampling threshold
- `presence_penalty`: Topic diversity control
- `frequency_penalty`: Repetition control

Note: No defaults are sent unless explicitly set. Invalid parameters for the selected model will be handled/ignored by the server.

**Usage:**
```python
# Configure for consistent, professional responses
agent.set_prompt_llm_params(
    temperature=0.3,
    top_p=0.9,
    presence_penalty=0.1,
    frequency_penalty=0.2
)
```

##### `set_post_prompt_llm_params`

Set Language Model parameters for the post-prompt. Accepts any parameters which will be passed through to the SignalWire server. The server validates and applies parameters based on the target model's capabilities.

<!-- snippet: no-compile signature-illustration -->
```python
def set_post_prompt_llm_params(**params) -> AgentBase
```

**Common Parameters:**
- `temperature`: Controls randomness. Lower = more focused
- `top_p`: Nucleus sampling threshold
- `presence_penalty`: Topic diversity control
- `frequency_penalty`: Repetition control

Note: No defaults are sent unless explicitly set.

**Usage:**
```python
# Configure for focused summaries
agent.set_post_prompt_llm_params(
    temperature=0.2,
    top_p=0.9
)
```

#### Structured Prompts (POM)

##### `prompt_add_section`

Add a structured section to the prompt using Prompt Object Model.

<!-- snippet: no-compile signature-illustration -->
```python
def prompt_add_section(
    title: str, 
    body: str = "", 
    bullets: Optional[List[str]] = None, 
    numbered: bool = False, 
    numbered_bullets: bool = False, 
    subsections: Optional[List[Dict[str, Any]]] = None
) -> AgentBase
```

**Parameters:**
- `title` (str): Section title/heading
- `body` (str): Main section content (default: "")
- `bullets` (Optional[List[str]]): List of bullet points
- `numbered` (bool): Use numbered sections (default: False)
- `numbered_bullets` (bool): Use numbered bullet points (default: False)
- `subsections` (Optional[List[Dict]]): Nested subsections

**Usage:**
```python
# Simple section
agent.prompt_add_section("Role", "You are a customer service representative.")

# Section with bullets
agent.prompt_add_section(
    "Guidelines", 
    "Follow these principles:",
    bullets=["Be helpful", "Stay professional", "Listen carefully"]
)

# Numbered bullets
agent.prompt_add_section(
    "Process",
    "Follow these steps:",
    bullets=["Greet the customer", "Identify their need", "Provide solution"],
    numbered_bullets=True
)
```

##### `prompt_add_to_section`

Add content to an existing prompt section.

<!-- snippet: no-compile signature-illustration -->
```python
def prompt_add_to_section(
    title: str, 
    body: Optional[str] = None, 
    bullet: Optional[str] = None, 
    bullets: Optional[List[str]] = None
) -> AgentBase
```

**Parameters:**
- `title` (str): Title of existing section to modify
- `body` (Optional[str]): Additional body text to append
- `bullet` (Optional[str]): Single bullet point to add
- `bullets` (Optional[List[str]]): Multiple bullet points to add

**Usage:**
```python
# Add body text to existing section
agent.prompt_add_to_section("Guidelines", "Remember to always verify customer identity.")

# Add single bullet
agent.prompt_add_to_section("Process", bullet="Document the interaction")

# Add multiple bullets
agent.prompt_add_to_section("Process", bullets=["Follow up", "Close ticket"])
```

##### `prompt_add_subsection`

Add a subsection to an existing prompt section.

<!-- snippet: no-compile signature-illustration -->
```python
def prompt_add_subsection(
    parent_title: str, 
    title: str, 
    body: str = "", 
    bullets: Optional[List[str]] = None
) -> AgentBase
```

**Parameters:**
- `parent_title` (str): Title of parent section
- `title` (str): Subsection title
- `body` (str): Subsection content (default: "")
- `bullets` (Optional[List[str]]): Subsection bullet points

**Usage:**
```python
agent.prompt_add_subsection(
    "Guidelines",
    "Escalation Rules", 
    "Escalate when:",
    bullets=["Customer is angry", "Technical issue beyond scope"]
)
```

### Voice and Language Configuration

##### `add_language`

Configure voice and language settings for the agent.

<!-- snippet: no-compile signature-illustration -->
```python
def add_language(
    name: str, 
    code: str, 
    voice: str, 
    speech_fillers: Optional[List[str]] = None, 
    function_fillers: Optional[List[str]] = None, 
    engine: Optional[str] = None, 
    model: Optional[str] = None
) -> AgentBase
```

**Parameters:**
- `name` (str): Human-readable language name
- `code` (str): Language code (e.g., "en-US", "es-ES")
- `voice` (str): Voice identifier (e.g., "rime.spore", "nova.luna")
- `speech_fillers` (Optional[List[str]]): Filler phrases during speech processing
- `function_fillers` (Optional[List[str]]): Filler phrases during function execution
- `engine` (Optional[str]): TTS engine to use
- `model` (Optional[str]): AI model to use

**Usage:**
```python
# Basic language setup
agent.add_language("English", "en-US", "rime.spore")

# With custom fillers
agent.add_language(
    "English", 
    "en-US", 
    "nova.luna",
    speech_fillers=["Let me think...", "One moment..."],
    function_fillers=["Processing...", "Looking that up..."]
)
```

##### `set_languages(languages: List[Dict[str, Any]]) -> AgentBase`
Set multiple language configurations at once.

**Parameters:**
- `languages` (List[Dict]): List of language configuration dictionaries

**Usage:**
```python
agent.set_languages([
    {"name": "English", "code": "en-US", "voice": "rime.spore"},
    {"name": "Spanish", "code": "es-ES", "voice": "nova.luna"}
])
```

### Speech Recognition Configuration

##### `add_hint(hint: str) -> AgentBase`
Add a single speech recognition hint.

**Parameters:**
- `hint` (str): Word or phrase to improve recognition accuracy

**Usage:**
```python
agent.add_hint("SignalWire")
```

##### `add_hints(hints: List[str]) -> AgentBase`
Add multiple speech recognition hints.

**Parameters:**
- `hints` (List[str]): List of words/phrases for better recognition

**Usage:**
```python
agent.add_hints(["SignalWire", "SWML", "API", "webhook", "SIP"])
```

##### `add_pattern_hint`

Add a pattern-based hint for speech recognition.

<!-- snippet: no-compile signature-illustration -->
```python
def add_pattern_hint(
    hint: str, 
    pattern: str, 
    replace: str, 
    ignore_case: bool = False
) -> AgentBase
```

**Parameters:**
- `hint` (str): The hint phrase
- `pattern` (str): Regex pattern to match
- `replace` (str): Replacement text
- `ignore_case` (bool): Case-insensitive matching (default: False)

**Usage:**
```python
agent.add_pattern_hint(
    "phone number",
    r"(\d{3})-(\d{3})-(\d{4})",
    r"(\1) \2-\3"
)
```

##### `add_pronunciation`

Add pronunciation rules for text-to-speech.

<!-- snippet: no-compile signature-illustration -->
```python
def add_pronunciation(
    replace: str, 
    with_text: str, 
    ignore_case: bool = False
) -> AgentBase
```

**Parameters:**
- `replace` (str): Text to replace
- `with_text` (str): Replacement pronunciation
- `ignore_case` (bool): Case-insensitive replacement (default: False)

**Usage:**
```python
agent.add_pronunciation("API", "A P I")
agent.add_pronunciation("SWML", "swim-el")
```

##### `set_pronunciations`

Set multiple pronunciation rules at once.

<!-- snippet: no-compile signature-illustration -->
```python
def set_pronunciations(
    pronunciations: List[Dict[str, Any]]
) -> AgentBase
```

**Parameters:**
- `pronunciations` (List[Dict]): List of pronunciation rule dictionaries

**Usage:**
```python
agent.set_pronunciations([
    {"replace": "API", "with": "A P I"},
    {"replace": "SWML", "with": "swim-el", "ignore_case": True}
])
```

### AI Parameters Configuration

##### `set_param(key: str, value: Any) -> AgentBase`
Set a single AI parameter.

**Parameters:**
- `key` (str): Parameter name
- `value` (Any): Parameter value

**Usage:**
```python
agent.set_param("ai_model", "gpt-4.1-nano")
agent.set_param("end_of_speech_timeout", 500)
```

##### `set_params(params: Dict[str, Any]) -> AgentBase`
Set multiple AI parameters at once.

**Parameters:**
- `params` (Dict[str, Any]): Dictionary of parameter key-value pairs

**Common Parameters:**
- `ai_model`: AI model to use ("gpt-4.1-nano", "gpt-4.1-mini", etc.)
- `end_of_speech_timeout`: Milliseconds to wait for speech end (default: 1000)
- `attention_timeout`: Milliseconds before attention timeout (default: 30000)
- `background_file_volume`: Volume for background audio (-60 to 0 dB)
- `temperature`: AI creativity/randomness (0.0 to 2.0)
- `max_tokens`: Maximum response length
- `top_p`: Nucleus sampling parameter (0.0 to 1.0)

**Usage:**
```python
agent.set_params({
    "ai_model": "gpt-4.1-nano",
    "end_of_speech_timeout": 500,
    "attention_timeout": 15000,
    "background_file_volume": -20,
    "temperature": 0.7
})
```

### Global Data Management

##### `set_global_data(data: Dict[str, Any]) -> AgentBase`
Set global data available to the AI and functions.

**Parameters:**
- `data` (Dict[str, Any]): Global data dictionary

**Usage:**
```python
agent.set_global_data({
    "company_name": "Acme Corp",
    "support_hours": "9 AM - 5 PM EST",
    "escalation_number": "+1-555-0123"
})
```

##### `update_global_data(data: Dict[str, Any]) -> AgentBase`
Update existing global data (merge with existing).

**Parameters:**
- `data` (Dict[str, Any]): Data to merge with existing global data

**Usage:**
```python
agent.update_global_data({
    "current_promotion": "20% off all services",
    "promotion_expires": "2024-12-31"
})
```

##### `set_semantic_gates(enabled: Optional[bool] = None, timeout_ms: Optional[int] = None, history: Optional[int] = None) -> AgentBase`
Set how the call checks its functions' [semantic gates](#semantic-gates). Each argument left as None keeps the platform's default.

**Parameters:**
- `enabled` (Optional[bool]): False dispatches gated functions without checking their gates. Default True
- `timeout_ms` (Optional[int]): The time one check may take, 500 to 10000 milliseconds. A check that runs out blocks the call. Default 2500
- `history` (Optional[int]): How many recent dialogue entries the decision model sees, 0 to 100. Default 20

**Raises:** `ValueError` for a value of the wrong type or out of range.

**Usage:**
```python
agent.set_semantic_gates(timeout_ms=4000, history=30)
```

### Function Definition

##### `define_tool`

Define a custom SWAIG function/tool.

<!-- snippet: no-compile signature-illustration -->
```python
def define_tool(
    name: str,
    description: str,
    parameters: Dict[str, Any],
    handler: Callable,
    secure: bool = True,
    fillers: Optional[Dict[str, List[str]]] = None,
    webhook_url: Optional[str] = None,
    is_typed_handler: bool = False,
    gates: Optional[List[Union[SemanticGate, Dict[str, Any]]]] = None,
    gate_fillers: Optional[Dict[str, List[Union[str, List[str]]]]] = None,
    **swaig_fields
) -> AgentBase
```

**Parameters:**
- `name` (str): Function name
- `description` (str): Function description for AI
- `parameters` (Dict[str, Any]): JSON schema for function parameters. If omitted when using the decorator and the handler has type-hinted parameters, the schema is inferred automatically from the type hints.
- `handler` (Callable): Function to execute when called
- `secure` (bool): Require security token (default: True)
- `fillers` (Optional[Dict[str, List[str]]]): Phrases the AI says while the function runs, keyed by language code, `auto` or `default`. The current language's own key is used; without one, the `auto` phrases are translated into that language on first use; without either, `default`. An entry may also be a list of phrases, a wait script, spoken one at a time while the call waits; the annotation predates wait scripts, so a type checker needs a cast for one
- `webhook_url` (Optional[str]): Custom webhook URL
- `gates` (Optional[List[Union[SemanticGate, Dict[str, Any]]]]): Semantic gates, 1 to 8. See [Semantic gates](#semantic-gates)
- `gate_fillers` (Optional[Dict[str, List[Union[str, List[str]]]]]): What the AI says while the gates are checked, shaped like `fillers`. Only with `gates`
- `**swaig_fields`: Additional SWAIG function properties

**Raises:** `ValueError` for a gate the platform would refuse, or `gate_fillers` without `gates`.

**Usage:**
```python
def get_weather(args, raw_data):
    location = args.get("location", "Unknown")
    return FunctionResult(f"The weather in {location} is sunny and 75°F")

agent.define_tool(
    name="get_weather",
    description="Get current weather for a location",
    parameters={
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "City name"
            }
        },
        "required": ["location"]
    },
    handler=get_weather,
    fillers={"en-US": ["Checking weather...", "Looking up forecast..."]}
)
```

##### Semantic gates

A semantic gate is a yes/no question a decision model answers about the call right before the platform dispatches the function. Every gate on the function must pass for it to run. When one doesn't, the function isn't dispatched, and the model gets that gate's `on_fail` output instead, as if a data_map had returned it. The function's `fillers` and `wait_file` start only once its gates pass, so a blocked call never sounds as if the action were under way.

Semantic gates need a platform release that supports them. One that doesn't ignores `gates` with a warning and runs the function ungated, so don't rely on a gate to protect a function until your platform supports them.

The decision model sees three things, under these names: `conversation`, the recent dialogue; `semantic_state`, the call's `global_data.semantic_state`, which `FunctionResult.set_semantic_state()` sets; and `proposed_function`, the function's name, description and arguments. Write each question as one proposition, and name what it's about with those names in backticks. The model reads literally: it doesn't do arithmetic or compare dates, so put a computed result in `semantic_state` instead of asking for it.

```python
from signalwire import FunctionResult, SemanticGate

agent.define_tool(
    name="cancel_account",
    description="Cancel the caller's account.",
    parameters={},
    handler=cancel_account,
    gates=[
        SemanticGate(
            "Has the caller explicitly asked to cancel their account in `conversation`?",
            threshold=0.95,
            on_fail=FunctionResult(
                tool_result="cancel_account was not run.",
                tool_prompt="Ask the caller to confirm that they want to cancel.",
            ).update_global_data({"cancel_attempted": True}),
            true_means="The caller says they want to cancel.",
            false_means="The caller asked about cancelling, or said something else.",
            id="explicit_request",
        )
    ],
    gate_fillers={"default": ["Let me verify that."]},
)
```

`SemanticGate(question, threshold, on_fail, *, id=None, true_means=None, false_means=None, visual_input=None)`:
- `question` (str): The yes/no question, at most 8 KB. Its `${...}` variables are expanded from the call's global data when the gate is checked
- `threshold` (float): The probability of yes, above 0 and at most 1, at or above which the gate passes
- `on_fail` (Union[str, FunctionResult, Dict[str, Any]]): What the model gets when this is the first gate that fails: the tool result text, a `FunctionResult`, whose response and actions are used, or a data_map output dict. It needs a response, and in the `tool_result`/`tool_prompt` form a `tool_result`: a blocked call must never read as a success. The `tool_prompt` becomes a system message after the tool result; OpenAI Realtime agents don't use it, so put what they need in the `tool_result`. The actions run as written, without template expansion
- `id` (Optional[str]): 1 to 64 letters, digits or underscores, unique within the function. Default `gate_<n>`, 1-based
- `true_means`, `false_means` (Optional[str]): What yes and no mean, at most 2 KB each
- `visual_input` (Optional[Union[bool, int]]): How many of the caller's newest video frames the decision model sees with this gate, as `visual_input`: True is 1, False or 0 none, or a whole number up to 5. Name it in the question, as in "Is the caller holding a photo ID up to the camera in `visual_input`?". All of a function's gates share one request, so each sees as many frames as the gate asking for the most. The gate provider must take images; one that doesn't makes the check a configuration error, which blocks the call. Left out by default. A function's own `visual_input`, frames sent to its webhook, is a separate `define_tool()` field

Gates are checked by the platform's rules when the tool is defined. The platform refuses the whole function for an invalid gate, so the SDK raises `ValueError` instead. Gates aren't allowed on the hook names `startup_hook`, `hangup_hook` and `check_for_input`, on `end_call`, or on a built-in function's name.

When a gate check can't be completed, because the decision model timed out or failed, the call is blocked too, and the model is told the action couldn't be completed right now. `set_semantic_gates()` sets the time a check may take.

##### `@AgentBase.tool(name=None, **kwargs)` (Class Decorator)
Decorator for defining tools as class methods.

**Parameters:**
- `name` (Optional[str]): Function name (defaults to method name)
- `**kwargs`: Same parameters as `define_tool()`

When `parameters` is omitted and the handler has type-hinted parameters (beyond `self`), the schema is inferred automatically from the type hints. The description is extracted from the docstring's first line, and per-parameter descriptions come from the `Args:` block.

**Usage (explicit schema):**
```python
class MyAgent(AgentBase):
    @AgentBase.tool(
        description="Get current time",
        parameters={"type": "object", "properties": {}}
    )
    def get_time(self, args, raw_data):
        import datetime
        return FunctionResult(f"Current time: {datetime.datetime.now()}")
```

**Usage (type-hinted, schema inferred):**
```python
class MyAgent(AgentBase):
    @AgentBase.tool(name="get_weather")
    def get_weather(self, city: str, units: str = "celsius"):
        """Get the weather forecast.

        Args:
            city: Name of the city
            units: Temperature units
        """
        return FunctionResult(f"Weather in {city}")
```

##### `register_swaig_function`

Register a pre-built SWAIG function dictionary.

<!-- snippet: no-compile signature-illustration -->
```python
def register_swaig_function(
    function_dict: Dict[str, Any]
) -> AgentBase
```

**Parameters:**
- `function_dict` (Dict[str, Any]): Complete SWAIG function definition

**Usage:**
```python
# Register a DataMap tool
weather_tool = DataMap('get_weather').webhook('GET', 'https://api.weather.com/...')
agent.register_swaig_function(weather_tool.to_swaig_function())
```

### Session Lifecycle Hooks

SignalWire AI agents support special SWAIG functions that are automatically called at specific points in the conversation lifecycle. `on_call_start()` and `on_call_end()` (see [Event Handlers](#event-handlers)) define them for you; to write one yourself, use the reserved name:

##### `startup_hook`
Called when a new conversation/call begins.

**Implementation:**
```python
@AgentBase.tool(
    name="startup_hook",
    description="Called when a new conversation starts to initialize state",
    parameters={}
)
def startup_hook(self, args, raw_data):
    call_id = raw_data.get("call_id")
    # Initialize session resources, load user data, etc.
    return FunctionResult("Session initialized")
```

##### `hangup_hook`
Called when a conversation/call ends.

**Implementation:**
```python
@AgentBase.tool(
    name="hangup_hook",
    description="Called when conversation ends to clean up resources",
    parameters={}
)
def hangup_hook(self, args, raw_data):
    call_id = raw_data.get("call_id")
    # Clean up resources, save session data, etc.
    return FunctionResult("Session ended")
```

**Common Use Cases:**
- Loading user preferences at session start
- Initializing session-specific resources
- Logging conversation metrics
- Cleaning up temporary data
- Saving conversation summaries

### Skills System

##### `add_skill`

Add a modular skill to the agent.

<!-- snippet: no-compile signature-illustration -->
```python
def add_skill(
    skill_name: str, 
    params: Optional[Dict[str, Any]] = None
) -> AgentBase
```

**Parameters:**
- `skill_name` (str): Name of the skill to add
- `params` (Optional[Dict[str, Any]]): Skill configuration parameters

**Available Skills:**
- `datetime`: Current date/time information
- `math`: Mathematical calculations
- `web_search`: Google Custom Search integration
- `datasphere`: SignalWire DataSphere search
- `native_vector_search`: Local document search

**Usage:**
```python
# Simple skill
agent.add_skill("datetime")
agent.add_skill("math")

# Skill with configuration
agent.add_skill("web_search", {
    "api_key": "your-google-api-key",
    "search_engine_id": "your-search-engine-id",
    "num_results": 3
})

# Multiple instances with different names
agent.add_skill("web_search", {
    "api_key": "your-api-key",
    "search_engine_id": "general-engine",
    "tool_name": "search_general"
})

agent.add_skill("web_search", {
    "api_key": "your-api-key", 
    "search_engine_id": "news-engine",
    "tool_name": "search_news"
})
```

##### `remove_skill(skill_name: str) -> AgentBase`
Remove a skill from the agent.

**Parameters:**
- `skill_name` (str): Name of skill to remove

**Usage:**
```python
agent.remove_skill("web_search")
```

##### `list_skills() -> List[str]`
Get list of currently added skills.

**Returns:**
- List[str]: Names of active skills

**Usage:**
```python
active_skills = agent.list_skills()
print(f"Active skills: {active_skills}")
```

##### `has_skill(skill_name: str) -> bool`
Check if a skill is currently added.

**Parameters:**
- `skill_name` (str): Name of skill to check

**Returns:**
- bool: True if skill is active

**Usage:**
```python
if agent.has_skill("web_search"):
    print("Web search is available")
```

### Native Functions

##### `set_native_functions`

Enable specific native SWML functions.

<!-- snippet: no-compile signature-illustration -->
```python
def set_native_functions(
    function_names: List[str]
) -> AgentBase
```

**Parameters:**
- `function_names` (List[str]): List of native function names to enable

**Available Native Functions:**
- `transfer`: Transfer calls
- `hangup`: End calls
- `play`: Play audio files
- `record`: Record audio
- `send_sms`: Send SMS messages

**Usage:**
```python
agent.set_native_functions(["transfer", "hangup", "send_sms"])
```

##### `set_internal_fillers`

Set custom filler phrases for internal/native SWAIG functions.

<!-- snippet: no-compile signature-illustration -->
```python
def set_internal_fillers(
    internal_fillers: Dict[str, Dict[str, List[str]]]
) -> AgentBase
```

**Parameters:**
- `internal_fillers` (Dict[str, Dict[str, List[str]]]): Maps a function name to a language code, which maps to filler phrases

**Available Internal Functions:**
- `next_step`: Moving between workflow steps (contexts system)
- `change_context`: Switching contexts in workflows  
- `check_time`: Getting current time
- `wait_for_user`: Waiting for user input
- `wait_seconds`: Pausing for specified duration
- `get_visual_input`: Processing visual data

**Usage:**
```python
agent.set_internal_fillers({
    "next_step": {
        "en-US": ["Moving to the next step...", "Let's continue..."],
        "es": ["Pasando al siguiente paso...", "Continuemos..."]
    },
    "check_time": {
        "en-US": ["Let me check the time...", "Getting current time..."]
    }
})
```

##### `add_internal_filler`

Add internal fillers for a specific function and language.

<!-- snippet: no-compile signature-illustration -->
```python
def add_internal_filler(
    function_name: str, 
    language_code: str, 
    fillers: List[str]
) -> AgentBase
```

**Parameters:**
- `function_name` (str): Name of the internal function
- `language_code` (str): Language code (e.g., "en-US", "es", "fr")
- `fillers` (List[str]): List of filler phrases

**Usage:**
```python
agent.add_internal_filler("next_step", "en-US", [
    "Great! Let's move to the next step...",
    "Perfect! Moving forward..."
])
```

### Function Includes

##### `add_function_include`

Include external SWAIG functions from another service.

<!-- snippet: no-compile signature-illustration -->
```python
def add_function_include(
    url: str, 
    functions: List[str], 
    meta_data: Optional[Dict[str, Any]] = None
) -> AgentBase
```

**Parameters:**
- `url` (str): URL of external SWAIG service
- `functions` (List[str]): List of function names to include
- `meta_data` (Optional[Dict[str, Any]]): Additional metadata

**Usage:**
```python
agent.add_function_include(
    "https://external-service.com/swaig",
    ["external_function1", "external_function2"],
    meta_data={"service": "external", "version": "1.0"}
)
```

##### `set_function_includes`

Set multiple function includes at once.

<!-- snippet: no-compile signature-illustration -->
```python
def set_function_includes(
    includes: List[Dict[str, Any]]
) -> AgentBase
```

**Parameters:**
- `includes` (List[Dict[str, Any]]): List of function include configurations

**Usage:**
```python
agent.set_function_includes([
    {
        "url": "https://service1.com/swaig",
        "functions": ["func1", "func2"]
    },
    {
        "url": "https://service2.com/swaig", 
        "functions": ["func3"],
        "meta_data": {"priority": "high"}
    }
])
```

### Webhook Configuration

##### `set_web_hook_url(url: str) -> AgentBase`
Set default webhook URL for SWAIG functions.

**Parameters:**
- `url` (str): Default webhook URL

**Usage:**
```python
agent.set_web_hook_url("https://myserver.com/webhook")
```

##### `set_post_prompt_url(url: str) -> AgentBase`
Set URL for post-prompt processing.

**Parameters:**
- `url` (str): Post-prompt webhook URL

**Usage:**
```python
agent.set_post_prompt_url("https://myserver.com/post-prompt")
```

##### `add_swaig_query_params(params: dict) -> AgentBase`
Add query parameters to be included in all SWAIG webhook URLs.

This is useful for preserving dynamic configuration state across SWAIG callbacks. For example, if your dynamic config adds skills based on query parameters, you can pass those same parameters through to the SWAIG webhook. This keeps the same configuration applied.

**Parameters:**
- `params` (dict): Dictionary of query parameter key-value pairs

**Usage:**
```python
# In dynamic config callback, preserve configuration parameters
def configure_agent(query_params, headers, body, agent):
    customer_id = query_params.get("customer_id")
    if customer_id:
        # Pass through to SWAIG callbacks
        agent.add_swaig_query_params({"customer_id": customer_id})
        agent.add_skill("customer_lookup", {"customer_id": customer_id})

agent.set_dynamic_config_callback(configure_agent)
```

##### `clear_swaig_query_params() -> AgentBase`
Clear all SWAIG query parameters.

**Usage:**
```python
agent.clear_swaig_query_params()
```

### Debug Events

##### `enable_debug_events`

Enable the debug event webhook for this agent. When enabled, the AI module will POST real-time debug events to a `/debug_events` endpoint on this agent during calls. Events are automatically logged via the agent's structured logger and can optionally be handled with a custom callback via `on_debug_event()`.

<!-- snippet: no-compile signature-illustration -->
```python
def enable_debug_events(level: int = 1) -> AgentBase
```

**Parameters:**
- `level` (int): Debug event verbosity level. `1` = high-level events (barge, errors, session start/end, step changes). `2+` = adds high-volume events (every LLM request/response, conversation_add). Default: `1`

**Usage:**
```python
agent.enable_debug_events()        # level 1 (default)
agent.enable_debug_events(level=2) # include high-volume events
```

**How it works:**
- Registers a `/debug_events` POST endpoint on the agent's HTTP server
- Auto-sets `debug_webhook_url` and `debug_webhook_level` in the SWML `params` during rendering
- The URL is built automatically using the same auth/proxy logic as other webhook URLs
- No manual URL configuration needed

**Event types at level 1:**

| Event label | Description |
|-------------|-------------|
| `session_start` | AI session started (model, TTS engine, voice, language) |
| `session_end` | AI session ended (reason, duration, token counts) |
| `barge` | User interrupted AI speech (barge type, elapsed ms) |
| `step_change` | Conversation step changed |
| `context_change` | Conversation context changed |
| `llm_error` | LLM error (fatal, retry, max_retries) |
| `voice_error` | TTS voice configuration or runtime error |
| `hold` | Call placed on hold or taken off hold |
| `filler` | Filler phrase spoken (thinking or function filler) |
| `consolidation` | Token consolidation triggered |
| `process_action` | Webhook action being processed |
| `gather_start` | Gather flow started |
| `gather_complete` | Gather flow completed |

**Additional events at level 2+:**

| Event label | Description |
|-------------|-------------|
| `llm_request` | LLM API request initiated (input tokens) |
| `llm_response` | LLM API response received (duration, output tokens) |
| `conversation_add` | Entry added to conversation history |

### Call Flow Verb Insertion

These methods allow you to customize the SWML call flow by inserting verbs at different stages of the call lifecycle.

##### `add_pre_answer_verb(verb_name: str, config: dict) -> AgentBase`
Add a verb to run before the call is answered (while still ringing).

**Safe pre-answer verbs:** `transfer`, `execute`, `return`, `label`, `goto`, `request`, `switch`, `cond`, `if`, `eval`, `set`, `unset`, `hangup`, `send_sms`, `sleep`, `stop_record_call`, `stop_denoise`, `stop_tap`

**Parameters:**
- `verb_name` (str): The SWML verb name
- `config` (dict): Verb configuration dictionary

**Usage:**
```python
# Send SMS before answering
agent.add_pre_answer_verb("send_sms", {
    "to": "+15551234567",
    "from": "+15559876543",
    "body": "Incoming call from AI agent"
})

# Set variables before answer
agent.add_pre_answer_verb("set", {"call_start": "${system.timestamp}"})
```

##### `add_answer_verb(config: dict = None) -> AgentBase`
Configure the answer verb that connects the call.

**Parameters:**
- `config` (dict, optional): Answer verb configuration (e.g., `{"max_duration": 3600}`)

**Usage:**
```python
# Set maximum call duration to 1 hour
agent.add_answer_verb({"max_duration": 3600})
```

##### `add_post_answer_verb(verb_name: str, config: dict) -> AgentBase`
Add a verb to run after the call is answered but before the AI starts.

**Parameters:**
- `verb_name` (str): The SWML verb name (e.g., "play", "sleep")
- `config` (dict): Verb configuration dictionary

**Usage:**
```python
# Play welcome message before AI starts
agent.add_post_answer_verb("play", {
    "url": "say:Welcome to our AI assistant. This call may be recorded."
})

# Add a brief pause
agent.add_post_answer_verb("sleep", {"duration": 1})
```

##### `add_post_ai_verb(verb_name: str, config: dict) -> AgentBase`
Add a verb to run after the AI conversation ends.

**Parameters:**
- `verb_name` (str): The SWML verb name (e.g., "hangup", "transfer", "request")
- `config` (dict): Verb configuration dictionary

**Usage:**
```python
# Clean hangup after AI ends
agent.add_post_ai_verb("hangup", {})

# Transfer to human after AI conversation
agent.add_post_ai_verb("transfer", {"to": "+15551234567"})

# Log call completion
agent.add_post_ai_verb("request", {
    "url": "https://myserver.com/call-complete",
    "method": "POST"
})
```

##### `clear_pre_answer_verbs() -> AgentBase`
Remove all pre-answer verbs.

##### `clear_post_answer_verbs() -> AgentBase`
Remove all post-answer verbs.

##### `clear_post_ai_verbs() -> AgentBase`
Remove all post-AI verbs.

**Method Chaining Example:**
```python
agent.add_pre_answer_verb("set", {"source": "ai_agent"}) \
     .add_answer_verb({"max_duration": 1800}) \
     .add_post_answer_verb("play", {"url": "say:Hello!"}) \
     .add_post_ai_verb("hangup", {})
```

### Dynamic Configuration

##### `set_dynamic_config_callback`

Set callback for per-request dynamic configuration.

<!-- snippet: no-compile signature-illustration -->
```python
def set_dynamic_config_callback(
    callback: Callable[[dict, dict, dict, AgentBase], None]
) -> AgentBase
```

**Parameters:**
- `callback` (Callable): Function that receives (query_params, headers, body, config)

**Usage:**
```python
def configure_agent(query_params, headers, body, config):
    # Configure based on request
    if query_params.get("language") == "spanish":
        config.add_language("Spanish", "es-ES", "nova.luna")
    
    # Set customer-specific data
    customer_id = headers.get("X-Customer-ID")
    if customer_id:
        config.set_global_data({"customer_id": customer_id})

agent.set_dynamic_config_callback(configure_agent)
```

### SIP Integration

##### `enable_sip_routing`

Enable SIP-based routing for voice calls.

<!-- snippet: no-compile signature-illustration -->
```python
def enable_sip_routing(
    auto_map: bool = True, 
    path: str = "/sip"
) -> AgentBase
```

**Parameters:**
- `auto_map` (bool): Automatically map SIP usernames (default: True)
- `path` (str): SIP routing endpoint path (default: "/sip")

**Usage:**
```python
agent.enable_sip_routing()
```

##### `register_sip_username(sip_username: str) -> AgentBase`
Register a specific SIP username for this agent.

**Parameters:**
- `sip_username` (str): SIP username to register

**Usage:**
```python
agent.register_sip_username("support")
agent.register_sip_username("sales")
```

##### `register_routing_callback`

Register custom routing logic for SIP calls.

<!-- snippet: no-compile signature-illustration -->
```python
def register_routing_callback(
    callback_fn: Callable[[Dict[str, Any], Dict[str, Any]], Optional[str]], 
    path: str = "/sip"
) -> None
```

**Parameters:**
- `callback_fn` (Callable): Function that receives the parsed body and headers, `(body, headers)`, and returns a route or `None`
- `path` (str): Routing endpoint path (default: "/sip")

**Usage:**
```python
def route_call(body, headers):
    sip_username = body.get("sip_username")
    if sip_username == "support":
        return "/support-agent"
    elif sip_username == "sales":
        return "/sales-agent"
    return None

agent.register_routing_callback(route_call)
```

### Utility Methods

##### `get_name() -> str`
Get the agent's name.

**Returns:**
- str: Agent name

##### `get_app()`
Get the FastAPI application instance.

**Returns:**
- FastAPI: The underlying FastAPI app

##### `as_router() -> APIRouter`
Get the agent as a FastAPI router for embedding in larger applications.

**Returns:**
- APIRouter: FastAPI router instance

**Usage:**
<!-- snippet: no-run FastAPI integration fragment (framework objects established in surrounding prose) -->
```python
# Embed agent in larger FastAPI app
main_app = FastAPI()
agent_router = agent.as_router()
main_app.include_router(agent_router, prefix="/agent")
```

### Event Handlers

##### `on_summary`

Override to handle conversation summaries. This callback is triggered when the AI generates a summary based on your `post_prompt` configuration.

<!-- snippet: no-compile signature-illustration -->
```python
def on_summary(
    summary: Optional[Dict[str, Any]],
    raw_data: Optional[Dict[str, Any]] = None
) -> None
```

**Parameters:**
- `summary` (Optional[Dict[str, Any]]): Parsed summary data (from `post_prompt_data.parsed[0]`)
- `raw_data` (Optional[Dict[str, Any]]): Complete raw POST data including `post_prompt_data` with both `raw` and `parsed` fields

**Usage:**
```python
class MyAgent(AgentBase):
    def __init__(self):
        super().__init__(name="summary-agent", route="/agent")

        # Configure post-prompt to request JSON summary
        self.set_post_prompt("""
        Return a JSON summary of the conversation:
        {
            "topic": "MAIN_TOPIC",
            "satisfied": true/false,
            "follow_up_needed": true/false,
            "key_points": ["point1", "point2"]
        }
        """)

    def on_summary(self, summary, raw_data):
        """Handle conversation summaries after call ends"""
        if summary:
            # Access parsed JSON fields directly
            topic = summary.get("topic", "Unknown")
            satisfied = summary.get("satisfied", False)

            print(f"Call about: {topic}, Customer satisfied: {satisfied}")

            # Save to database, send to CRM, trigger follow-up, etc.
            if summary.get("follow_up_needed"):
                self.schedule_follow_up(summary)

        # Access raw summary text if needed
        if raw_data and 'post_prompt_data' in raw_data:
            raw_text = raw_data['post_prompt_data'].get('raw', '')
            print(f"Raw summary: {raw_text}")
```

##### `on_call_start`

Register a handler that runs when the call starts, before the AI's first turn. Use as a decorator or call directly; handlers run in registration order. Wraps the platform's reserved `startup_hook` function.

<!-- snippet: no-compile signature-illustration -->
```python
def on_call_start(
    handler: Callable[[Dict[str, Any]], Optional[FunctionResult]]
) -> Callable[[Dict[str, Any]], Optional[FunctionResult]]
```

The handler receives `raw_data`, the SWAIG request (`call_id`, `global_data`, `caller_id_num` and the rest), and may return a `FunctionResult` or `None`:
- Its actions run at once, in registration order across handlers. `post_process` isn't used.
- Its response text is added to the conversation as a system message the model reads before it speaks. Several handlers' texts are joined, one per line. Only a plain string response is used. An OpenAI Realtime agent runs the actions but drops the text, so put what its first greeting needs in the prompt from a per-call configuration callback instead.

An exception is logged, not raised, and that handler's result is dropped. The platform waits for the hook before the AI starts, so keep the handler short.

**Usage:**
```python
agent = AgentBase("my_agent")

@agent.on_call_start
def greet_by_name(raw_data):
    name = raw_data.get("caller_id_name") or "the caller"
    return FunctionResult(f"The caller's name is {name}.")
```

##### `on_call_end`

Register a handler that runs when the call ends, with the conversation. Use as a decorator or call directly; handlers run in registration order. Wraps the platform's reserved `hangup_hook` function, and turns on `swaig_post_conversation` so the hook carries the transcript.

<!-- snippet: no-compile signature-illustration -->
```python
def on_call_end(
    handler: Callable[[List[Dict[str, Any]], Dict[str, Any]], None]
) -> Callable[[List[Dict[str, Any]], Dict[str, Any]], None]
```

The handler receives `call_log` (the conversation) and `raw_data` (the SWAIG request). Its return value is ignored, and an exception is logged, not raised.

**Usage:**
```python
agent = AgentBase("my_agent")

@agent.on_call_end
def log_length(call_log, raw_data):
    print(f"Call {raw_data.get('call_id')} ended after {len(call_log)} entries")
```

##### `on_debug_event`

Register a handler for debug webhook events. Use as a decorator. Requires `enable_debug_events()` to be called first.

<!-- snippet: no-compile signature-illustration -->
```python
def on_debug_event(handler: Callable) -> Callable
```

The handler receives:
- `event_type` (str): The event label (e.g. `"barge"`, `"llm_error"`, `"session_start"`)
- `data` (dict): The full event payload including `call_id`, `label`, and event-specific fields

The handler may be sync or async.

**Usage (decorator style):**
```python
agent = AgentBase("my_agent")
agent.enable_debug_events()

@agent.on_debug_event
def handle_debug(event_type, data):
    call_id = data.get("call_id")
    if event_type == "llm_error":
        print(f"LLM error on call {call_id}: {data.get('event')}")
    elif event_type == "barge":
        print(f"Barge after {data.get('barge_elapsed_ms')}ms")
    elif event_type == "session_end":
        print(f"Call ended: {data.get('reason')}, duration: {data.get('duration_ms')}ms")
```

**Usage (subclass style):**
```python
class MyAgent(AgentBase):
    def __init__(self):
        super().__init__(name="debug-agent", route="/agent")
        self.enable_debug_events(level=2)
        self.on_debug_event(self.handle_debug)

    def handle_debug(self, event_type, data):
        if event_type == "llm_error":
            self.alert_ops_team(data)
```

> **Note:** Even without registering a handler, all debug events are automatically logged via the agent's structured logger when `enable_debug_events()` is called.

##### `on_function_call`

Override to handle function calls with custom logic.

<!-- snippet: no-compile signature-illustration -->
```python
def on_function_call(
    name: str,
    args: Dict[str, Any],
    raw_data: Optional[Dict[str, Any]] = None
) -> Any
```

**Parameters:**
- `name` (str): Function name being called
- `args` (Dict[str, Any]): Function arguments
- `raw_data` (Optional[Dict[str, Any]]): Raw request data

**Returns:**
- Any: Function result (typically FunctionResult)

**Usage:**
```python
class MyAgent(AgentBase):
    def on_function_call(self, name, args, raw_data):
        if name == "get_weather":
            location = args.get("location")
            # Custom weather logic
            return FunctionResult(f"Weather in {location}: Sunny")
        return super().on_function_call(name, args, raw_data)
```

##### `on_request`

Override to handle general requests.

<!-- snippet: no-compile signature-illustration -->
```python
def on_request(
    request_data: Optional[dict] = None, 
    callback_path: Optional[str] = None
) -> Optional[dict]
```

**Parameters:**
- `request_data` (Optional[dict]): Request data
- `callback_path` (Optional[str]): Callback path

**Returns:**
- Optional[dict]: Response modifications

##### `on_swml_request`

Override to handle SWML generation requests.

<!-- snippet: no-compile signature-illustration -->
```python
def on_swml_request(
    request_data: Optional[dict] = None, 
    callback_path: Optional[str] = None, 
    request: Optional[Request] = None
) -> Optional[dict]
```

**Parameters:**
- `request_data` (Optional[dict]): Request data
- `callback_path` (Optional[str]): Callback path  
- `request` (Optional[Request]): FastAPI request object

**Returns:**
- Optional[dict]: SWML modifications

### Authentication

##### `validate_basic_auth(username: str, password: str) -> bool`
Override to implement custom basic authentication logic.

**Parameters:**
- `username` (str): Username from basic auth
- `password` (str): Password from basic auth

**Returns:**
- bool: True if credentials are valid

**Usage:**
```python
class MyAgent(AgentBase):
    def validate_basic_auth(self, username, password):
        # Custom auth logic
        return username == "admin" and password == "secret"
```

##### `get_basic_auth_credentials`

Get basic auth credentials from environment or constructor.

<!-- snippet: no-compile signature-illustration -->
```python
def get_basic_auth_credentials(
    include_source: bool = False
) -> Union[Tuple[str, str], Tuple[str, str, str]]
```

**Parameters:**
- `include_source` (bool): Include source information (default: False)

**Returns:**
- Tuple: (username, password) or (username, password, source)

### Context System

##### `define_contexts() -> ContextBuilder`
Define structured workflow contexts for the agent.

**Returns:**
- ContextBuilder: Builder for creating contexts and steps

**Usage:**
```python
contexts = agent.define_contexts()

greeting = contexts.add_context("greeting")
greeting.add_step("welcome", task="Welcome! How can I help?",
                  valid_steps=["main_menu"])

main_menu = contexts.add_context("main_menu")
main_menu.add_step("menu", task="Choose: 1) Support 2) Sales 3) Billing",
                   functions=["transfer_to_support", "transfer_to_sales"])
```

---

## FunctionResult Class

The `FunctionResult` class is used to create structured responses from SWAIG functions. It handles both natural language responses and structured actions that the agent should execute.

### Constructor

This is the full constructor signature:

<!-- snippet: no-compile signature-illustration -->
```python
FunctionResult(
    response: Optional[str] = None, 
    post_process: bool = False
)
```

**Parameters:**
- `response` (Optional[str]): Natural language response text for the AI to speak
- `post_process` (bool): Whether to let AI take another turn before executing actions (default: False)

**Post-processing Behavior:**
- `post_process=False` (default): Execute actions immediately after AI response
- `post_process=True`: Let AI respond to user one more time, then execute actions

**Usage:**
```python
# Simple response
result = FunctionResult("The weather is sunny and 75°F")

# Response with post-processing enabled
result = FunctionResult("I'll transfer you now", post_process=True)

# Empty response (actions only)
result = FunctionResult()
```

### Core Methods

#### Response Configuration

##### `set_response(response: str) -> FunctionResult`
Set or update the natural language response text.

**Parameters:**
- `response` (str): The text the AI should speak

**Usage:**
```python
result = FunctionResult()
result.set_response("I found your order information")
```

##### `set_post_process(post_process: bool) -> FunctionResult`
Enable or disable post-processing for this result.

**Parameters:**
- `post_process` (bool): True to let AI respond once more before executing actions

**Usage:**
```python
result = FunctionResult("I'll help you with that")
result.set_post_process(True)  # Let AI handle follow-up questions first
```

#### Action Management

##### `add_action(name: str, data: Any) -> FunctionResult`
Add a structured action to execute.

**Parameters:**
- `name` (str): Action name/type (e.g., "play", "transfer", "set_global_data")
- `data` (Any): Action data - can be string, boolean, object, or array

**Usage:**
```python
# Simple action with boolean
result.add_action("hangup", True)

# Action with string data
result.add_action("play", "welcome.mp3")

# Action with object data
result.add_action("set_global_data", {"customer_id": "12345", "status": "verified"})

# Action with array data
result.add_action("send_sms", ["+15551234567", "Your order is ready!"])
```

##### `add_actions(actions: List[Dict[str, Any]]) -> FunctionResult`
Add multiple actions at once.

**Parameters:**
- `actions` (List[Dict[str, Any]]): List of action dictionaries

**Usage:**
```python
result.add_actions([
    {"play": "hold_music.mp3"},
    {"set_global_data": {"status": "on_hold"}},
    {"wait": 5000}
])
```

### Call Control Actions

#### Call Transfer and Connection

##### `connect(destination: str, final: bool = True, from_addr: Optional[str] = None) -> FunctionResult`
Transfer or connect the call to another destination.

**Parameters:**
- `destination` (str): Phone number, SIP address, or other destination
- `final` (bool): Permanent transfer (True) vs temporary transfer (False) (default: True)
- `from_addr` (Optional[str]): Override caller ID

**Transfer Types:**
- `final=True`: Permanent transfer - call exits agent completely
- `final=False`: Temporary transfer - call returns to agent if far end hangs up

**Usage:**
```python
# Permanent transfer to phone number
result.connect("+15551234567", final=True)

# Temporary transfer to SIP address with custom caller ID
result.connect("support@company.com", final=False, from_addr="+15559876543")

# Transfer with response
result = FunctionResult("Transferring you to our sales team")
result.connect("sales@company.com")
```

##### `swml_transfer(dest: str, ai_response: str, final: bool = True) -> FunctionResult`
Create a SWML-based transfer with AI response setup.

**Parameters:**
- `dest` (str): Transfer destination
- `ai_response` (str): AI response when transfer completes
- `final` (bool): Whether this is a permanent transfer (True) or temporary (False). Defaults to True.

**Usage:**
```python
result.swml_transfer(
    "+15551234567", 
    "You've been transferred back to me. How else can I help?"
)
```

##### `sip_refer(to_uri: str) -> FunctionResult`
Perform a SIP REFER transfer.

**Parameters:**
- `to_uri` (str): SIP URI to transfer to

**Usage:**
```python
result.sip_refer("sip:support@company.com")
```

#### Call Management

##### `hangup() -> FunctionResult`
End the call immediately.

**Usage:**
```python
result = FunctionResult("Thank you for calling. Goodbye!")
result.hangup()
```

##### `hold(prompt: Optional[Union[str, int]] = None, timeout: int = 300, step: Optional[str] = None, timeout_step: Optional[str] = None) -> FunctionResult`
Put the call on hold. Speech detection pauses for the duration of the hold, so say anything the caller needs to hear before it takes effect. Pass `prompt` to have the agent say it for you automatically.

**Parameters:**
- `prompt` (Optional[str]): Instruction the model speaks before the hold takes effect. Setting it also turns on `post_process`, so the model gets one more turn to speak before the `hold` action runs. A single positional `int` here is read as `timeout` instead, so `hold(120)` still means `hold(timeout=120)`.
- `timeout` (int): Seconds to hold, clamped to a maximum of 900 (default: 300)
- `step` (Optional[str]): Step to enter when the call comes off hold normally, before the timeout
- `timeout_step` (Optional[str]): Step to enter when the hold reaches its `timeout` with nobody releasing it. Without it, a timed-out hold resumes in the step it left.

Both transitions are deferred: they fire only when the hold actually ends. Returning `swml_change_step()` alongside `hold()` moves the caller immediately, before the hold even begins, which is a separate, earlier transition.

**Usage:**
```python
result = FunctionResult("Please hold while I look that up")
result.hold(timeout=60)
```

Announce the hold and route the caller when it ends:

```python
result = FunctionResult().hold(
    prompt="Tell the caller you are checking whether someone is available.",
    timeout=60,
    step="human_available",
    timeout_step="take_message",
)
```

##### `stop() -> FunctionResult`
Stop the agent's execution.

**Usage:**
```python
result.stop()
```

#### Audio Control

##### `say(text: str) -> FunctionResult`
Add text for the AI to speak.

**Parameters:**
- `text` (str): Text to speak

**Usage:**
```python
result.say("Please wait while I process your request")
```

##### `play_background_file(filename: str, wait: bool = False) -> FunctionResult`
Play an audio file in the background.

**Parameters:**
- `filename` (str): Audio file path or URL
- `wait` (bool): Suppress the agent's attention-getting behavior during playback (default: False)

**Usage:**
```python
# Play hold music in background
result.play_background_file("hold_music.mp3")

# Play an announcement without the agent trying to get attention
result.play_background_file("important_announcement.wav", wait=True)
```

##### `stop_background_file() -> FunctionResult`
Stop background audio playback.

**Usage:**
```python
result.stop_background_file()
```

### Data Management Actions

##### `update_global_data(data: Dict[str, Any]) -> FunctionResult`
Update existing global data (merge with existing).

**Parameters:**
- `data` (Dict[str, Any]): Data to merge

**Usage:**
```python
result.update_global_data({
    "last_interaction": "2024-01-15T10:30:00Z",
    "agent_notes": "Customer satisfied with resolution"
})
```

##### `remove_global_data(keys: Union[str, List[str]]) -> FunctionResult`
Remove specific keys from global data.

**Parameters:**
- `keys` (Union[str, List[str]]): Key name or list of key names to remove

**Usage:**
```python
# Remove single key
result.remove_global_data("temporary_data")

# Remove multiple keys
result.remove_global_data(["temp1", "temp2", "cache_data"])
```

##### `set_semantic_state(state: Dict[str, Any]) -> FunctionResult`
Set the state the call's [semantic gates](#semantic-gates) judge, besides the dialogue: `global_data.semantic_state`. It replaces the state as a whole: to change one field, send the whole state with that field changed, and send `{}` to reset it. Keep it to what the application has established, such as an order the caller confirmed, rather than what the caller claims. Nothing else from the global data reaches the decision model.

**Parameters:**
- `state` (Dict[str, Any]): The semantic state

**Usage:**
```python
result.set_semantic_state({"order": {"item": "large pepperoni", "confirmed": True}})
```

##### `set_metadata(data: Dict[str, Any]) -> FunctionResult`
Set metadata for the conversation.

**Parameters:**
- `data` (Dict[str, Any]): Metadata to set

**Usage:**
```python
result.set_metadata({
    "call_type": "support",
    "priority": "high",
    "department": "technical"
})
```

##### `remove_metadata(keys: Union[str, List[str]]) -> FunctionResult`
Remove specific metadata keys.

**Parameters:**
- `keys` (Union[str, List[str]]): Key name or list of key names to remove

**Usage:**
```python
result.remove_metadata(["temporary_flag", "debug_info"])
```

### AI Behavior Control

##### `change_voice(voice: str) -> FunctionResult`
Change the AI's voice for the rest of the call. The new voice replaces the current language's voice from the next batch of speech on, never mid-utterance, and stays for that language for the rest of the call. It may be on another engine than the current one. A voice that won't open falls back to the fallback voice.

**Parameters:**
- `voice` (str): The voice as `engine.voice:model`, the form a language's voice takes

**Raises:** `ValueError` if `voice` is empty.

**Usage:**
```python
result.change_voice("elevenlabs.rachel")
```

##### `set_end_of_speech_timeout(milliseconds: int) -> FunctionResult`
Adjust how long to wait for speech to end.

**Parameters:**
- `milliseconds` (int): Timeout in milliseconds

**Usage:**
```python
# Shorter timeout for quick responses
result.set_end_of_speech_timeout(300)

# Longer timeout for thoughtful responses
result.set_end_of_speech_timeout(2000)
```

##### `set_speech_event_timeout(milliseconds: int) -> FunctionResult`
Set timeout for speech events.

**Parameters:**
- `milliseconds` (int): Timeout in milliseconds

**Usage:**
```python
result.set_speech_event_timeout(5000)
```

##### `wait_for_user(enabled: Optional[bool] = None, timeout: Optional[int] = None, answer_first: bool = False) -> FunctionResult`
Control whether to wait for user input.

**Parameters:**
- `enabled` (Optional[bool]): Enable/disable waiting for user
- `timeout` (Optional[int]): Timeout in seconds
- `answer_first` (bool): Answer call before waiting (default: False)

**Usage:**
```python
# Wait for user input with 10 second timeout
result.wait_for_user(enabled=True, timeout=10)

# Don't wait for user (immediate response)
result.wait_for_user(enabled=False)
```

##### `toggle_functions(function_toggles: List[Dict[str, Any]]) -> FunctionResult`
Enable or disable specific functions.

**Parameters:**
- `function_toggles` (List[Dict[str, Any]]): List of function toggle configurations

**Usage:**
```python
result.toggle_functions([
    {"function": "transfer_to_sales", "active": True},
    {"function": "end_call", "active": False}
])
```

##### `enable_functions_on_timeout(enabled: bool = True) -> FunctionResult`
Control whether functions are enabled when timeout occurs.

**Parameters:**
- `enabled` (bool): Enable functions on timeout (default: True)

**Usage:**
```python
result.enable_functions_on_timeout(False)  # Disable functions on timeout
```

##### `enable_extensive_data(enabled: bool = True) -> FunctionResult`
Enable extensive data collection.

**Parameters:**
- `enabled` (bool): Enable extensive data (default: True)

**Usage:**
```python
result.enable_extensive_data(True)
```

##### `update_settings(settings: Dict[str, Any]) -> FunctionResult`
Update various AI settings.

**Parameters:**
- `settings` (Dict[str, Any]): Settings to update

**Usage:**
```python
result.update_settings({
    "temperature": 0.8,
    "max_tokens": 150,
    "end_of_speech_timeout": 800
})
```

### Context and Conversation Control

##### `switch_context(system_prompt: Optional[str] = None, user_prompt: Optional[str] = None, consolidate: bool = False, full_reset: bool = False) -> FunctionResult`
Switch conversation context or reset the conversation.

**Parameters:**
- `system_prompt` (Optional[str]): New system prompt
- `user_prompt` (Optional[str]): New user prompt
- `consolidate` (bool): Consolidate conversation history (default: False)
- `full_reset` (bool): Completely reset conversation (default: False)

**Usage:**
```python
# Switch to technical support context
result.switch_context(
    system_prompt="You are now a technical support specialist",
    user_prompt="The customer needs technical help"
)

# Reset conversation completely
result.switch_context(full_reset=True)

# Consolidate conversation history
result.switch_context(consolidate=True)
```

##### `simulate_user_input(text: str) -> FunctionResult`
Simulate user input for testing or automation.

**Parameters:**
- `text` (str): Text to simulate as user input

**Usage:**
```python
result.simulate_user_input("I need help with my order")
```

### Communication Actions

##### `send_sms(to_number: str, from_number: str, body: Optional[str] = None, media: Optional[List[str]] = None, tags: Optional[List[str]] = None, region: Optional[str] = None) -> FunctionResult`
Send an SMS message.

**Parameters:**
- `to_number` (str): Recipient phone number
- `from_number` (str): Sender phone number
- `body` (Optional[str]): SMS message text
- `media` (Optional[List[str]]): List of media URLs
- `tags` (Optional[List[str]]): Message tags
- `region` (Optional[str]): SignalWire region

**Usage:**
```python
# Simple text message
result.send_sms(
    to_number="+15551234567",
    from_number="+15559876543", 
    body="Your order #12345 has shipped!"
)

# Message with media and tags
result.send_sms(
    to_number="+15551234567",
    from_number="+15559876543",
    body="Here's your receipt",
    media=["https://example.com/receipt.pdf"],
    tags=["receipt", "order_12345"]
)
```

### Recording and Media

##### `record_call(control_id: Optional[str] = None, stereo: bool = False, format: str = "wav", direction: str = "both", terminators: Optional[str] = None, beep: bool = False, input_sensitivity: float = 44.0, initial_timeout: Optional[float] = None, end_silence_timeout: Optional[float] = None, max_length: Optional[float] = None, status_url: Optional[str] = None) -> FunctionResult`
Start call recording.

**Parameters:**
- `control_id` (Optional[str]): Unique identifier for this recording
- `stereo` (bool): Record in stereo (default: False)
- `format` (str): Recording format: "wav", "mp3", "mp4" (default: "wav")
- `direction` (str): Recording direction: "speak", "listen", "both" (default: "both")
- `terminators` (Optional[str]): DTMF keys to stop recording
- `beep` (bool): Play beep before recording (default: False)
- `input_sensitivity` (float): Input sensitivity level (default: 44.0)
- `initial_timeout` (Optional[float]): Time to wait for speech start, in seconds (default: unset)
- `end_silence_timeout` (Optional[float]): Time to wait in silence before ending, in seconds (default: unset)
- `max_length` (Optional[float]): Maximum recording length in seconds
- `status_url` (Optional[str]): Webhook URL for recording status

**Usage:**
```python
# Basic recording
result.record_call(format="mp3", direction="both")

# Recording with control ID and settings
result.record_call(
    control_id="customer_call_001",
    stereo=True,
    format="wav",
    beep=True,
    max_length=300.0,
    terminators="#*"
)
```

##### `stop_record_call(control_id: Optional[str] = None) -> FunctionResult`
Stop call recording.

**Parameters:**
- `control_id` (Optional[str]): Control ID of recording to stop

**Usage:**
```python
result.stop_record_call()
result.stop_record_call(control_id="customer_call_001")
```

### Conference and Room Management

##### `join_room(name: str) -> FunctionResult`
Join a SignalWire room.

**Parameters:**
- `name` (str): Room name to join

**Usage:**
```python
result.join_room("support_room_1")
```

##### `join_conference(name: str, muted: bool = False, beep: str = "true", start_on_enter: bool = True, end_on_exit: bool = False, wait_url: Optional[str] = None, max_participants: Optional[int] = None, record: str = "do-not-record", region: Optional[str] = None, trim: str = "trim-silence", coach: Optional[str] = None, status_callback_event: Optional[str] = None, status_callback: Optional[str] = None, status_callback_method: str = "POST", recording_status_callback: Optional[str] = None, recording_status_callback_method: str = "POST", recording_status_callback_event: str = "completed", result: Optional[Any] = None) -> FunctionResult`
Join a conference call.

**Parameters:**
- `name` (str): Conference name
- `muted` (bool): Join muted (default: False)
- `beep` (str): Beep setting: "true", "false", "onEnter", "onExit" (default: "true")
- `start_on_enter` (bool): Start conference when this participant enters (default: True)
- `end_on_exit` (bool): End conference when this participant exits (default: False)
- `wait_url` (Optional[str]): URL for hold music/content
- `max_participants` (int): Maximum participants, 2 or more (default: None, which leaves it out so the platform's default applies)
- `record` (str): Recording setting (default: "do-not-record")
- `region` (Optional[str]): SignalWire region
- `trim` (str): Trim setting for recordings (default: "trim-silence")
- `coach` (Optional[str]): Coach participant identifier
- `status_callback_event` (Optional[str]): Status callback events
- `status_callback` (Optional[str]): Status callback URL
- `status_callback_method` (str): Status callback HTTP method (default: "POST")
- `recording_status_callback` (Optional[str]): Recording status callback URL
- `recording_status_callback_method` (str): Recording status callback method (default: "POST")
- `recording_status_callback_event` (str): Recording status callback events (default: "completed")

**Usage:**
```python
# Basic conference join
result.join_conference("sales_meeting")

# Conference with recording and settings
result.join_conference(
    name="support_conference",
    muted=False,
    beep="onEnter",
    record="record-from-start",
    max_participants=10
)
```

### Payment Processing

##### `pay(payment_connector_url: str, input_method: str = "dtmf", status_url: Optional[str] = None, payment_method: str = "credit-card", timeout: int = 5, max_attempts: int = 1, security_code: bool = True, postal_code: Union[bool, str] = True, min_postal_code_length: int = 0, token_type: str = "reusable", charge_amount: Optional[str] = None, currency: str = "usd", language: str = "en-US", voice: str = "woman", description: Optional[str] = None, valid_card_types: str = "visa mastercard amex", parameters: Optional[List[Dict[str, str]]] = None, prompts: Optional[List[Dict[str, Any]]] = None) -> FunctionResult`
Process a payment through the call.

**Parameters:**
- `payment_connector_url` (str): Payment processor webhook URL
- `input_method` (str): only "dtmf" is valid; the SWML pay verb's schema requires this constant (default: "dtmf")
- `status_url` (Optional[str]): Payment status webhook URL
- `payment_method` (str): Payment method: "credit-card" (default: "credit-card")
- `timeout` (int): Input timeout in seconds (default: 5)
- `max_attempts` (int): Maximum retry attempts (default: 1)
- `security_code` (bool): Require security code (default: True)
- `postal_code` (Union[bool, str]): Require postal code (default: True)
- `min_postal_code_length` (int): Minimum postal code length (default: 0)
- `token_type` (str): Token type: "reusable", "one-time" (default: "reusable")
- `charge_amount` (Optional[str]): Amount to charge
- `currency` (str): Currency code (default: "usd")
- `language` (str): Language for prompts (default: "en-US")
- `voice` (str): Voice for prompts (default: "woman")
- `description` (Optional[str]): Payment description
- `valid_card_types` (str): Accepted card types (default: "visa mastercard amex")
- `parameters` (Optional[List[Dict[str, str]]]): Additional parameters
- `prompts` (Optional[List[Dict[str, Any]]]): Custom prompts

**Usage:**
```python
# Basic payment processing
result.pay(
    payment_connector_url="https://payment-processor.com/webhook",
    charge_amount="29.99",
    description="Monthly subscription"
)

# Payment with custom settings
result.pay(
    payment_connector_url="https://payment-processor.com/webhook",
    timeout=10,
    max_attempts=3,
    security_code=True,
    postal_code=True,
    charge_amount="149.99",
    currency="usd",
    description="Premium service upgrade"
)
```

### Call Monitoring

##### `tap(uri: str, control_id: Optional[str] = None, direction: str = "both", codec: str = "PCMU", rtp_ptime: int = 20, status_url: Optional[str] = None) -> FunctionResult`
Start call tapping/monitoring.

**Parameters:**
- `uri` (str): URI to send tapped audio to: `rtp://IP:port`, `ws://example.com`, or `wss://example.com`
- `control_id` (Optional[str]): Unique identifier for this tap
- `direction` (str): Tap direction: "speak", "listen", "both" (default: "both")
- `codec` (str): Audio codec: "PCMU" or "PCMA" (default: "PCMU")
- `rtp_ptime` (int): RTP packet time in milliseconds (default: 20)
- `status_url` (Optional[str]): Status webhook URL

**Usage:**
```python
# Basic call tapping
result.tap("wss://monitor.company.com/tap")

# Tap with specific settings
result.tap(
    uri="rtp://192.168.1.100:5004",
    control_id="quality_monitor_001",
    direction="both",
    codec="PCMA"
)
```

##### `stop_tap(control_id: Optional[str] = None) -> FunctionResult`
Stop call tapping.

**Parameters:**
- `control_id` (Optional[str]): Control ID of tap to stop

**Usage:**
```python
result.stop_tap()
result.stop_tap(control_id="quality_monitor_001")
```

### Advanced SWML Execution

##### `execute_swml(swml_content, transfer: bool = False) -> FunctionResult`
Execute custom SWML content.

**Parameters:**
- `swml_content`: SWML document or content to execute
- `transfer` (bool): Whether this is a transfer operation (default: False)

**Usage:**
```python
# Execute custom SWML
custom_swml = {
    "version": "1.0.0",
    "sections": {
        "main": [
            {"play": {"url": "https://example.com/custom.mp3"}},
            {"say": {"text": "Custom SWML execution"}}
        ]
    }
}
result.execute_swml(custom_swml)
```

### Utility Methods

##### `to_dict() -> Dict[str, Any]`
Convert the result to a dictionary for serialization.

**Returns:**
- Dict[str, Any]: Dictionary representation of the result

**Usage:**
```python
result = FunctionResult("Hello world")
result.add_action("play", "music.mp3")
result_dict = result.to_dict()
print(result_dict)
# Output: {"response": "Hello world", "action": [{"play": "music.mp3"}]}
```

### Static Helper Methods

##### `create_payment_prompt(for_situation: str, actions: List[Dict[str, str]], card_type: Optional[str] = None, error_type: Optional[str] = None) -> Dict[str, Any]`
Create a payment prompt configuration.

**Parameters:**
- `for_situation` (str): Situation identifier
- `actions` (List[Dict[str, str]]): List of action configurations
- `card_type` (Optional[str]): Card type for prompts
- `error_type` (Optional[str]): Error type for error prompts

**Usage:**
```python
prompt = FunctionResult.create_payment_prompt(
    for_situation="card_number",
    actions=[
        FunctionResult.create_payment_action("say", "Please enter your card number")
    ]
)
```

##### `create_payment_action(action_type: str, phrase: str) -> Dict[str, str]`
Create a payment action configuration.

**Parameters:**
- `action_type` (str): Action type
- `phrase` (str): Action phrase

**Usage:**
```python
action = FunctionResult.create_payment_action("say", "Enter your card number")
```

##### `create_payment_parameter(name: str, value: str) -> Dict[str, str]`
Create a payment parameter configuration.

**Parameters:**
- `name` (str): Parameter name
- `value` (str): Parameter value

**Usage:**
```python
param = FunctionResult.create_payment_parameter("merchant_id", "12345")
```

### Method Chaining

All methods return `self`, enabling fluent method chaining:

```python
result = (FunctionResult("I'll help you with that")
    .set_post_process(True)
    .update_global_data({"status": "helping"})
    .set_end_of_speech_timeout(800)
    .add_action("play", "thinking.mp3"))

# Complex workflow
result = (FunctionResult("Processing your payment")
    .set_post_process(True)
    .update_global_data({"payment_status": "processing"})
    .pay(
        payment_connector_url="https://payments.com/webhook",
        charge_amount="99.99",
        description="Service payment"
    )
    .send_sms(
        to_number="+15551234567",
        from_number="+15559876543",
        body="Payment confirmation will be sent shortly"
    ))
```

---

## DataMap Class

The `DataMap` class provides a declarative approach to creating SWAIG tools that integrate with REST APIs without requiring webhook infrastructure. DataMap tools execute on SignalWire's server infrastructure, eliminating the need to expose webhook endpoints.

### Constructor

`DataMap` takes the function name it will create:

<!-- snippet: no-compile signature-illustration -->
```python
DataMap(function_name: str)
```

**Parameters:**
- `function_name` (str): Name of the SWAIG function this DataMap will create

**Usage:**
```python
# Create a new DataMap tool
weather_map = DataMap('get_weather')
search_map = DataMap('search_docs')
```

### Core Configuration Methods

#### Function Metadata

##### `purpose(description: str) -> DataMap`
Set the function description/purpose.

**Parameters:**
- `description` (str): Human-readable description of what this function does

**Usage:**
```python
data_map = DataMap('get_weather').purpose('Get current weather information for any city')
```

##### `description(description: str) -> DataMap`
Alias for `purpose()` - set the function description.

**Parameters:**
- `description` (str): Function description

**Usage:**
```python
data_map = DataMap('search_api').description('Search our knowledge base for information')
```

#### Parameter Definition

##### `parameter(name: str, param_type: str, description: str, required: bool = False, enum: Optional[List[str]] = None) -> DataMap`
Add a function parameter with JSON schema validation.

**Parameters:**
- `name` (str): Parameter name
- `param_type` (str): JSON schema type: "string", "number", "boolean", "array", "object"
- `description` (str): Parameter description for the AI
- `required` (bool): Whether parameter is required (default: False)
- `enum` (Optional[List[str]]): List of allowed values for validation

**Usage:**
```python
# Required string parameter
data_map.parameter('location', 'string', 'City name or ZIP code', required=True)

# Optional number parameter
data_map.parameter('days', 'number', 'Number of forecast days', required=False)

# Enum parameter with allowed values
data_map.parameter('units', 'string', 'Temperature units', 
                  enum=['celsius', 'fahrenheit'], required=False)

# Boolean parameter
data_map.parameter('include_alerts', 'boolean', 'Include weather alerts', required=False)

# Array parameter
data_map.parameter('categories', 'array', 'Search categories to include')
```

### API Integration Methods

#### HTTP Webhook Configuration

##### `webhook(method: str, url: str, headers: Optional[Dict[str, str]] = None, form_param: Optional[str] = None, input_args_as_params: bool = False, require_args: Optional[List[str]] = None) -> DataMap`
Configure an HTTP API call.

**Parameters:**
- `method` (str): "GET" or "POST". The platform sends a POST when this is "POST" or the webhook has params, and a GET otherwise, so "PUT", "PATCH" and "DELETE" are sent as GET
- `url` (str): API endpoint URL (supports `${variable}` substitution)
- `headers` (Optional[Dict[str, str]]): HTTP headers to send, as written. The platform doesn't expand templates in header values
- `form_param` (Optional[str]): Send the JSON params as a single form field with this name
- `input_args_as_params` (bool): Merge the function's arguments into the params, the JSON body, which makes the request a POST (default: False)
- `require_args` (Optional[List[str]]): Skip this webhook, without a request, unless at least one of these arguments is present

The platform requests the first webhook whose `require_args` are met, and no other. When that webhook fails, it doesn't try the next one; the `fallback_output()` runs.

**Variable Substitution in URLs:**
- `${args.parameter_name}`: Function argument values
- `${global_data.key}`: Call-wide data store (user info, call state - NOT credentials)
- `${meta_data.key}`: The function's metadata
- `${call_id}`: The call's ID; other call details, such as `${caller_id_num}`, are at the root too

**Usage:**
```python
# Simple GET request with parameter substitution
data_map.webhook('GET', 'https://api.weather.com/v1/current?key=API_KEY&q=${args.location}')

# POST request with authentication headers
data_map.webhook(
    'POST', 
    'https://api.company.com/search',
    headers={
        'Authorization': 'Bearer YOUR_TOKEN',
        'Content-Type': 'application/json'
    }
)

# Webhook that requires specific arguments
data_map.webhook(
    'GET',
    'https://api.service.com/data?id=${args.customer_id}',
    require_args=['customer_id']
)

# Use global data for call-related info (NOT credentials)
data_map.webhook(
    'GET',
    'https://api.service.com/customer/${global_data.customer_id}/orders',
    headers={'Authorization': 'Bearer YOUR_API_TOKEN'}  # Use static credentials
)
```

##### `body(data: Dict[str, Any]) -> DataMap`
Set the JSON request body; the same as `params()`. The platform reads a webhook's body from its `params` field, so `body()` sets `params`, and the webhook is sent as a POST.

**Parameters:**
- `data` (Dict[str, Any]): JSON body data (supports `${variable}` substitution)

**Usage:**
```python
# Static body with parameter substitution
data_map.body({
    'query': '${args.search_term}',
    'limit': 5,
    'filters': {
        'category': '${args.category}',
        'active': True
    }
})

# Body with call-related data (NOT sensitive info)
data_map.body({
    'customer_id': '${global_data.customer_id}',
    'request_id': '${call_id}',
    'search': '${args.query}'
})
```

##### `params(data: Dict[str, Any]) -> DataMap`
Set the request's JSON body. The platform sends `params` as the body, not as URL query parameters, so a webhook with params is sent as a POST whatever its method. For a GET, put query parameters in the URL, as in `'https://api.example.com/weather?q=${enc:args.location}'`.

**Parameters:**
- `data` (Dict[str, Any]): Body fields (supports `${variable}` substitution against the call data, such as `${args.location}`)

**Usage:**
```python
# A JSON body with substitution
data_map.params({
    'api_key': 'YOUR_API_KEY',
    'q': '${args.location}',
    'units': '${args.units}',
    'lang': 'en'
})
```

#### Multiple Webhooks and Fallbacks

The platform requests one webhook per call: the first whose `require_args` are met. It doesn't try a later webhook when that one fails; the `fallback_output()` runs instead. Several webhooks are useful when `require_args` choose between them:

```python
# Look an order up by its number, or else by the customer's email
data_map = (DataMap('find_order')
    .purpose('Find an order by its number or the customer email')
    .parameter('order_id', 'string', 'Order number')
    .parameter('email', 'string', 'Customer email address')

    # Requested when the AI supplied an order number
    .webhook('GET', 'https://api.example.com/orders/${enc:args.order_id}', require_args=['order_id'])
    .output(FunctionResult('Order ${id}: ${status}'))

    # Requested when it supplied only an email
    .webhook('GET', 'https://api.example.com/orders?email=${enc:args.email}', require_args=['email'])
    .output(FunctionResult('Latest order for ${input.args.email}: ${array[0].status}'))

    # Used when the webhook that ran fails, or neither argument was supplied
    .fallback_output(FunctionResult('Sorry, the order service is unavailable right now'))
)
```

### Response Processing

#### Basic Output

##### `output(result: FunctionResult) -> DataMap`
Set the response template for successful API calls.

**Parameters:**
- `result` (FunctionResult): Response template with variable substitution

**Variable Substitution in Outputs:**
- `${field}`: fields of the API's JSON response, read from the root with no prefix
- `${nested.field}`: nested response fields
- `${array[0].field}`: elements of a response that is a JSON array
- `${input.args.parameter}`: Original function arguments. The call data is under `input` in an output, so `${args.parameter}` is empty there
- `${input.call_id}`, `${input.meta_data.key}`: the call details and the function's metadata
- `${global_data.key}`: Call-wide data store (user info, call state)
- `${prompt_vars.key}`: the prompt variables

**Usage:**
```python
# Simple response template
data_map.output(FunctionResult('Weather in ${input.args.location}: ${current.condition.text}, ${current.temp_f}°F'))

# Response with actions
data_map.output(
    FunctionResult('Found ${total_results} results')
    .update_global_data({'last_search': '${input.args.query}'})
    .add_action('play', 'search_complete.mp3')
)

# Complex response with nested data
data_map.output(
    FunctionResult('Order ${order.id} status: ${order.status}. Estimated delivery: ${order.delivery.estimated_date}')
)
```

##### `fallback_output(result: FunctionResult) -> DataMap`
Set the top-level output. The platform uses it when no expression matched and the webhook stage produced no result: the webhook it requested failed, or no webhook was eligible. Its templates read the call data, where an argument is `${args.x}`.

**Parameters:**
- `result` (FunctionResult): Fallback response

**Usage:**
```python
data_map.fallback_output(
    FunctionResult('Sorry, the service is temporarily unavailable. Please try again later.')
    .add_action('play', 'service_unavailable.mp3')
)
```

#### Array Processing

##### `foreach(foreach_config: Dict[str, Any]) -> DataMap`
Process array responses by iterating over elements.

**Parameters:**
- `foreach_config` (Dict): Configuration object with keys `input_key`, `output_key`, `append`, and optional `max`

**Simple Array Processing:**
```python
# Process array of search results
data_map = (DataMap('search_docs')
    .webhook('GET', 'https://api.docs.com/search?q=${args.query}')
    .foreach({
        'input_key': 'results',              # Key in the response holding the array
        'output_key': 'formatted_results',   # Name of the built string variable
        'append': 'Found: ${this.title} - ${this.summary}\n'
    })
    .output(FunctionResult('${formatted_results}'))
)
```

**Advanced Array Processing:**
```python
# Complex foreach configuration
data_map.foreach({
    'input_key': 'items',
    'output_key': 'formatted_items',
    'max': 3,  # Process only first 3 items
    'append': 'Item: ${this.name} (${this.status})\n'
})
```

**Foreach Variable Access:**
- `${this.field}`: Current array element field
- `${this.nested.field}`: Nested fields in current element
- `${output_key}`: The accumulated string built by `append` across all items

### Pattern-Based Processing

#### Expression Matching

##### `expression(test_value: str, pattern: Union[str, Pattern], output: FunctionResult, nomatch_output: Optional[FunctionResult] = None) -> DataMap`
Add pattern-based responses without API calls.

**Parameters:**
- `test_value` (str): Template string to test against pattern
- `pattern` (Union[str, Pattern]): Regex pattern or compiled Pattern object
- `output` (FunctionResult): Response when pattern matches
- `nomatch_output` (Optional[FunctionResult]): Response when pattern doesn't match

**Usage:**
```python
# Command-based responses
control_map = (DataMap('file_control')
    .purpose('Control file playback')
    .parameter('command', 'string', 'Playback command', required=True)
    .parameter('filename', 'string', 'File to control')
    .parameter('level', 'string', 'Volume level, from 0 to 10')

    # Start commands
    .expression(
        '${args.command}', 
        r'start|play|begin',
        FunctionResult('Starting playback')
        .add_action('start_playback', {'file': '${args.filename}'})
    )
    
    # Stop commands
    .expression(
        '${args.command}',
        r'stop|pause|halt',
        FunctionResult('Stopping playback')
        .add_action('stop_playback', True)
    )
    
    # Volume commands, with the level in its own argument
    .expression(
        '${args.command}',
        r'volume',
        FunctionResult('Setting volume to ${args.level}')
        .add_action('set_volume', '${args.level}')
    )
)
```

**Pattern Matching:**
The platform expands the test value against the call data and searches it for the pattern, a regular expression that can match anywhere in the value, without regard to case unless it's written as `/pattern/flags`. It doesn't expose capture groups to templates, so a value the output needs should come from its own argument, as `level` does here. Expressions run in order, and the first that produces an output ends the function.

### Error Handling

##### `error_keys(keys: List[str]) -> DataMap`
Specify response fields that indicate errors, for the most recent webhook. The webhook fails when its JSON response has any of these fields at the top level, whatever the value, and the `fallback_output()` runs. An HTTP status outside 200-299 isn't a failure by itself: the platform adds an `http_code` field to such a response, so list `http_code` to fail on it.

**Parameters:**
- `keys` (List[str]): List of field names that indicate API errors

**Usage:**
```python
# Treat these response fields as errors
data_map.error_keys(['error', 'error_message', 'http_code'])

# If the API returns {"error": "Not found"}, or any status outside 200-299, the webhook fails
```

##### `global_error_keys(keys: List[str]) -> DataMap`
Set a top-level `error_keys` field. The platform ignores it, and checks only each webhook's own error keys, so call `error_keys()` after the webhook instead. `error_keys()` called before any webhook sets the same ignored field.

**Parameters:**
- `keys` (List[str]): Global error field names

**Usage:**
```python
data_map.global_error_keys(['error', 'message', 'code'])
```

##### `gate(gate: Union[SemanticGate, Dict[str, Any]]) -> DataMap`
Add a [semantic gate](#semantic-gates), a yes/no precondition a decision model checks right before the platform runs the function. Call once per gate, up to 8. When a gate fails, the data_map doesn't run, and the model gets that gate's `on_fail` output. `to_swaig_function()` raises `ValueError` for a gate the platform would refuse.

**Usage:**
```python
data_map.gate(SemanticGate(
    "Has the caller confirmed the order in `conversation`?",
    threshold=0.9,
    on_fail="place_order was not run.",
))
```

##### `gate_fillers(fillers: Dict[str, List[Union[str, List[str]]]]) -> DataMap`
Set what the AI says while the function's gates are checked, shaped like a function's fillers. Only for a function with gates.

**Usage:**
```python
data_map.gate_fillers({"default": ["Let me check that."]})
```

### Advanced Configuration

##### `webhook_expressions(expressions: List[Dict[str, Any]]) -> DataMap`
Attach expressions to the most recently added webhook, storing them under its `expressions` field. Call `webhook()` first; calling this before any webhook has been added raises `ValueError`. They run after the webhook's `foreach`, against its response, and the first that produces an output replaces the webhook's own output. Their templates read the response's fields from the root, and the call data under `input`, such as `${input.args.id}`.

**Parameters:**
- `expressions` (List[Dict[str, Any]]): Expression objects to store as given, each needing the same `string`, `pattern` and `output` keys as `expression()`

**Usage:**
```python
# Evaluate the webhook's response and pick a reply based on its status field
data_map.webhook('GET', 'https://api.example.com/status?id=${args.id}')
data_map.webhook_expressions([
    {
        'string': '${status}',
        'pattern': 'complete',
        'output': FunctionResult('The task is complete.').to_dict()
    },
    {
        'string': '${status}',
        'pattern': 'pending',
        'output': FunctionResult('The task is still in progress.').to_dict()
    }
])
```

### Complete DataMap Examples

#### Simple Weather API

This tool calls a weather API directly and formats the response:

```python
weather_tool = (DataMap('get_weather')
    .purpose('Get current weather information')
    .parameter('location', 'string', 'City name or ZIP code', required=True)
    .parameter('units', 'string', 'Temperature units', enum=['celsius', 'fahrenheit'])
    .webhook('GET', 'https://api.weather.com/v1/current?key=API_KEY&q=${enc:args.location}&units=${enc:args.units}')
    .output(FunctionResult('Weather in ${input.args.location}: ${current.condition.text}, ${current.temp_f}°F'))
    .error_keys(['error'])
)

# Register with agent
agent.register_swaig_function(weather_tool.to_swaig_function())
```

#### Search with Array Processing

This tool posts a query, then formats each result in the response array:

```python
search_tool = (DataMap('search_knowledge')
    .purpose('Search company knowledge base')
    .parameter('query', 'string', 'Search query', required=True)
    .parameter('category', 'string', 'Search category', enum=['docs', 'faq', 'policies'])
    .webhook(
        'POST', 
        'https://api.company.com/search',
        headers={'Authorization': 'Bearer TOKEN'}
    )
    .params({
        'query': '${args.query}',
        'category': '${args.category}',
        'limit': 5
    })
    .foreach({
        'input_key': 'results',
        'output_key': 'formatted_results',
        'append': 'Found: ${this.title} - ${this.summary}\n'
    })
    .output(FunctionResult('${formatted_results}'))
    .fallback_output(FunctionResult('Search service is temporarily unavailable'))
)
```

#### Command Processing (No API)

This tool matches the action against patterns and responds without calling an API:

```python
control_tool = (DataMap('system_control')
    .purpose('Control system functions')
    .parameter('action', 'string', 'Action to perform', required=True)
    .parameter('target', 'string', 'Target for the action')
    
    # Restart commands
    .expression(
        '${args.action}',
        r'restart|reboot',
        FunctionResult('Restarting ${args.target}')
        .add_action('restart_service', {'service': '${args.target}'})
    )
    
    # Status commands
    .expression(
        '${args.action}',
        r'status|check',
        FunctionResult('Checking status of ${args.target}')
        .add_action('check_status', {'service': '${args.target}'})
    )
    
    # Default for unrecognized commands
    .expression(
        '${args.action}',
        r'.*',
        FunctionResult('Unknown command: ${args.action}'),
        nomatch_output=FunctionResult('Please specify a valid action')
    )
)
```

### Conversion and Registration

##### `to_swaig_function() -> Dict[str, Any]`
Convert the DataMap to a SWAIG function dictionary for registration.

**Returns:**
- Dict[str, Any]: Complete SWAIG function definition

**Usage:**
```python
# Build DataMap
weather_map = DataMap('get_weather').purpose('Get weather').parameter('location', 'string', 'City', required=True)

# Convert to SWAIG function and register
swaig_function = weather_map.to_swaig_function()
agent.register_swaig_function(swaig_function)
```

### Convenience Functions

The SDK provides helper functions for common DataMap patterns:

##### `create_simple_api_tool(name: str, url: str, response_template: str, parameters: Optional[Dict[str, Dict]] = None, method: str = "GET", headers: Optional[Dict[str, str]] = None, body: Optional[Dict[str, Any]] = None, error_keys: Optional[List[str]] = None) -> DataMap`

Create a simple API integration tool.

**Parameters:**
- `name` (str): Function name
- `url` (str): API endpoint URL
- `response_template` (str): Response template string
- `parameters` (Optional[Dict[str, Dict]]): Parameter definitions
- `method` (str): HTTP method (default: "GET")
- `headers` (Optional[Dict[str, str]]): HTTP headers
- `body` (Optional[Dict[str, Any]]): JSON request body, set as the webhook's params
- `error_keys` (Optional[List[str]]): Error field names

**Usage:**
```python
from signalwire.core.data_map import create_simple_api_tool

weather = create_simple_api_tool(
    name='get_weather',
    url='https://api.weather.com/v1/current?key=API_KEY&q=${enc:args.location}',
    response_template='Weather in ${input.args.location}: ${current.condition.text}',
    parameters={
        'location': {
            'type': 'string', 
            'description': 'City name', 
            'required': True
        }
    }
)

agent.register_swaig_function(weather.to_swaig_function())
```

##### `create_expression_tool(name: str, patterns: Dict[str, Tuple[str, FunctionResult]], parameters: Optional[Dict[str, Dict]] = None) -> DataMap`

Create a pattern-based tool without API calls.

**Parameters:**
- `name` (str): Function name
- `patterns` (Dict[str, Tuple[str, FunctionResult]]): Pattern mappings
- `parameters` (Optional[Dict[str, Dict]]): Parameter definitions

**Usage:**
```python
from signalwire.core.data_map import create_expression_tool

file_control = create_expression_tool(
    name='file_control',
    # Maps a test value to a (pattern, FunctionResult) pair
    patterns={
        '${args.command}': (r'start|play', FunctionResult('Starting playback').add_action('start_playback', {'file': '${args.filename}'})),
    },
    parameters={
        'command': {
            'type': 'string',
            'description': 'Playback command',
            'required': True
        },
        'filename': {
            'type': 'string',
            'description': 'File to play'
        }
    }
)

agent.register_swaig_function(file_control.to_swaig_function())
```

A dict holds one pattern per test value. For several patterns on the same value, chain `DataMap.expression()` calls instead.

### Method Chaining

All DataMap methods return `self`, enabling fluent method chaining:

```python
complete_tool = (DataMap('comprehensive_search')
    .purpose('Search the knowledge base')
    .parameter('query', 'string', 'Search query', required=True)
    .parameter('category', 'string', 'Search category', enum=['all', 'docs', 'faq'])
    .webhook('GET', 'https://api.example.com/search?q=${enc:args.query}&cat=${enc:args.category}')
    .output(FunctionResult('Top result: ${title}'))
    .error_keys(['error', 'message'])
    .fallback_output(FunctionResult('Search is unavailable right now'))
)
```

---

## Context System

The Context System enhances traditional prompt-based agents by adding structured workflows with sequential steps on top of a base prompt. Each step contains its own guidance, completion criteria, and function restrictions while building upon the agent's foundational prompt.

### ContextBuilder Class

The `ContextBuilder` is accessed via `agent.define_contexts()` and provides the main interface for creating structured workflows.

#### Getting Started

This creates a context, then chains a step onto it:

```python
# Access the context builder
contexts = agent.define_contexts()

# Create contexts and steps
contexts.add_context("greeting") \
    .add_step("welcome") \
    .set_text("Welcome! How can I help you today?") \
    .set_step_criteria("User has stated their need") \
    .set_valid_steps(["next"])
```

##### `add_context(name: str) -> Context`
Create a new context in the workflow.

**Parameters:**
- `name` (str): Unique context name

**Returns:**
- Context: Context object for method chaining

**Usage:**
<!-- snippet: no-run illustrative fragment (references `contexts` established in the surrounding prose) -->
```python
# Create multiple contexts
greeting_context = contexts.add_context("greeting")
main_menu_context = contexts.add_context("main_menu")
support_context = contexts.add_context("support")
```

### Context Class

The Context class represents a conversation context containing multiple steps with enhanced features:

<!-- snippet: no-compile signature-illustration -->
```python
class Context:
    def add_step(self, name: str) -> Step
        """Create a new step in this context"""
    
    def set_valid_contexts(self, contexts: List[str]) -> Context
        """Set which contexts can be accessed from this context"""
        
    # Context entry parameters (for context switching behavior)
    def set_post_prompt(self, post_prompt: str) -> Context
        """Override agent's post prompt when this context is active"""
    
    def set_system_prompt(self, system_prompt: str) -> Context
        """Trigger context switch with new system instructions (makes this a Context Switch Context)"""
        
    def set_consolidate(self, consolidate: bool) -> Context
        """Whether to consolidate conversation history when entering this context"""
        
    def set_full_reset(self, full_reset: bool) -> Context
        """Whether to do complete system prompt replacement vs injection"""
        
    def set_user_prompt(self, user_prompt: str) -> Context
        """User message to inject when entering this context for AI context"""
    
    # Context prompts (guidance for all steps in context)
    def set_prompt(self, prompt: str) -> Context
        """Set simple string prompt that applies to all steps in this context"""
        
    def add_section(self, title: str, body: str) -> Context
        """Add POM-style section to context prompt"""
        
    def add_bullets(self, title: str, bullets: List[str]) -> Context
        """Add POM-style bullet section to context prompt"""
```

**Context Types:**

1. **Workflow Container Context** (no `system_prompt`): Organizes steps without conversation state changes
2. **Context Switch Context** (has `system_prompt`): Triggers conversation state changes when entered, processing entry parameters like a `context_switch` SWAIG action

**Prompt Hierarchy:** Base Agent Prompt, then Context Prompt, then Step Prompt

#### Usage Examples

Both context types are created the same way, and differ in whether `set_system_prompt` is called:

<!-- snippet: no-run illustrative fragment (references `contexts` established in the surrounding prose) -->
```python
# Workflow container context (just organizes steps)
main_context = contexts.add_context("main")
main_context.set_prompt("Follow standard customer service protocols")

# Context switch context (changes AI behavior)  
billing_context = contexts.add_context("billing")
billing_context.set_system_prompt("You are now a billing specialist") \
    .set_consolidate(True) \
    .set_user_prompt("Customer needs billing assistance") \
    .add_section("Department", "Billing Department") \
    .add_bullets("Services", ["Account inquiries", "Payments", "Refunds"])

# Full reset context (complete conversation reset)
manager_context = contexts.add_context("manager") 
manager_context.set_system_prompt("You are a senior manager") \
    .set_full_reset(True) \
    .set_consolidate(True)
```

---

## Skills System

The Skills System provides modular, reusable capabilities you can add to any agent.

### Available Built-in Skills

#### `datetime` Skill
Provides current date and time information. Exposes `get_current_time` and `get_current_date` tools; each accepts a `timezone` argument at call time (defaults to UTC).

**Parameters:** None (the skill takes no configuration parameters beyond the common base parameters).

**Usage:**
```python
# Add the datetime skill (no configuration parameters)
agent.add_skill("datetime")
```

#### `math` Skill
Safe mathematical expression evaluation. Exposes a `calculate` tool that evaluates an `expression` argument (supports +, -, *, /, %, ** and parentheses).

**Parameters:** None (the skill takes no configuration parameters beyond the common base parameters).

**Usage:**
```python
# Add the math skill (no configuration parameters)
agent.add_skill("math")
```

#### `web_search` Skill
Google Custom Search API integration with web scraping.

**Parameters:**
- `api_key` (str): Google Custom Search API key (required)
- `search_engine_id` (str): Google Custom Search Engine ID (required)
- `num_results` (Optional[int]): Number of results to return (default: 3)
- `tool_name` (Optional[str]): Custom tool name for multiple instances
- `delay` (Optional[float]): Delay between requests in seconds (default: 0.5)
- `no_results_message` (Optional[str]): Custom message when no results found

**Usage:**
```python
# Basic web search
agent.add_skill("web_search", {
    "api_key": "your-google-api-key",
    "search_engine_id": "your-search-engine-id"
})

# Multiple search instances
agent.add_skill("web_search", {
    "api_key": "your-api-key",
    "search_engine_id": "general-engine-id",
    "tool_name": "search_general",
    "num_results": 5
})

agent.add_skill("web_search", {
    "api_key": "your-api-key",
    "search_engine_id": "news-engine-id",
    "tool_name": "search_news",
    "num_results": 3,
    "delay": 0.5
})
```

#### `datasphere` Skill
SignalWire DataSphere knowledge search integration.

**Parameters:**
- `space_name` (str): DataSphere space name (required)
- `project_id` (str): DataSphere project ID (required)
- `token` (str): DataSphere access token (required)
- `document_id` (Optional[str]): Specific document to search
- `tool_name` (Optional[str]): Custom tool name for multiple instances
- `count` (Optional[int]): Number of results to return (default: 1)
- `tags` (Optional[List[str]]): Filter by document tags

**Usage:**
```python
# Basic DataSphere search
agent.add_skill("datasphere", {
    "space_name": "my-space",
    "project_id": "my-project",
    "token": "my-token"
})

# Multiple DataSphere instances
agent.add_skill("datasphere", {
    "space_name": "my-space",
    "project_id": "my-project",
    "token": "my-token",
    "document_id": "drinks-menu",
    "tool_name": "search_drinks",
    "count": 5
})

agent.add_skill("datasphere", {
    "space_name": "my-space",
    "project_id": "my-project", 
    "token": "my-token",
    "tool_name": "search_policies",
    "tags": ["HR", "Policies"]
})
```

#### `native_vector_search` Skill
Local document search with vector similarity and keyword search.

**Parameters:**
- `index_file` (str): Path to search index file (required)
- `tool_name` (Optional[str]): Custom tool name (default: "search_knowledge")
- `count` (Optional[int]): Maximum results to return (default: 5)
- `similarity_threshold` (Optional[float]): Minimum similarity score 0.0-1.0 (default: 0.0). Higher values are stricter, lower values are more permissive. Typical range: 0.2-0.4 for all-MiniLM-L6-v2, 0.3-0.5 for all-mpnet-base-v2
- `tags` (Optional[List[str]]): Filter results by document tags

**Usage:**
```python
# Basic local search
agent.add_skill("native_vector_search", {
    "index_file": "./knowledge.swsearch"
})

# With custom settings
agent.add_skill("native_vector_search", {
    "index_file": "./docs.swsearch",
    "tool_name": "search_docs",
    "count": 10,
    "similarity_threshold": 0.25
})
```

### Creating Custom Skills

#### Skill Structure

Create a new skill by extending `SkillBase`:

```python
from typing import List, Dict, Any
from signalwire.core.skill_base import SkillBase
from signalwire.core.data_map import DataMap
from signalwire.core.function_result import FunctionResult

class CustomSkill(SkillBase):
    SKILL_NAME = "custom_skill"
    SKILL_DESCRIPTION = "Description of what this skill does"
    SKILL_VERSION = "1.0.0"
    REQUIRED_PACKAGES = ["requests"]  # Python packages needed
    REQUIRED_ENV_VARS = ["API_KEY"]   # Environment variables needed
    
    def setup(self) -> bool:
        """Validate and store configuration"""
        if not self.params.get("api_key"):
            self.logger.error("api_key parameter is required")
            return False
        
        self.api_key = self.params["api_key"]
        return True
    
    def register_tools(self) -> None:
        """Register skill functions"""
        # DataMap-based tool
        tool = (DataMap("custom_function")
            .description("Custom API integration")
            .parameter("query", "string", "Search query", required=True)
            .webhook("GET", f"https://api.example.com/search?key={self.api_key}&q=${{args.query}}")
            .output(FunctionResult("Found: ${{title}}"))
        )
        
        self.agent.register_swaig_function(tool.to_swaig_function())
    
    def get_hints(self) -> List[str]:
        """Speech recognition hints"""
        return ["custom search", "find information"]
    
    def get_global_data(self) -> Dict[str, Any]:
        """Global data for DataMap"""
        return {"skill_version": self.SKILL_VERSION}
    
    def get_prompt_sections(self) -> List[Dict[str, Any]]:
        """Prompt sections to add"""
        return [{
            "title": "Custom Search Capability",
            "body": "You can search our custom database for information.",
            "bullets": ["Use the custom_function to search", "Results are real-time"]
        }]
```

#### Skill Registration

Skills are automatically discovered from the `signalwire/skills/` directory. To register a custom skill:

1. Create directory: `signalwire/skills/your_skill/`
2. Add `__init__.py`, `skill.py`, and `README.md`
3. Implement your skill class in `skill.py`
4. The skill will be automatically available

---

## Utility Classes

### SWAIGFunction Class

Represents a SWAIG function definition with metadata and validation.

#### Constructor

This is the full constructor signature:

<!-- snippet: no-compile signature-illustration -->
```python
SWAIGFunction(
    name: str,
    handler: Callable,
    description: str,
    parameters: Optional[Dict[str, Any]] = None,
    secure: bool = False,
    **kwargs
)
```

**Parameters:**
- `name` (str): Function name
- `handler` (Callable): Function to execute when called
- `description` (str): Function description
- `parameters` (Dict[str, Any]): JSON schema for parameters
- `secure` (bool): Require security token (default: False)
- `**kwargs`: Additional SWAIG properties

#### Usage

This builds the same function `define_tool()` would, one level lower:

```python
from signalwire.core.swaig_function import SWAIGFunction

# Create SWAIG function
def get_weather(args, raw_data):
    return f"The weather in {args.get('location')} is sunny."

swaig_func = SWAIGFunction(
    name="get_weather",
    handler=get_weather,
    description="Get current weather",
    parameters={
        "location": {"type": "string", "description": "City name"}
    },
    required=["location"],
    secure=True,
    fillers={"en-US": ["Checking weather..."]}
)

# Register with agent
agent.register_swaig_function(swaig_func.to_dict())
```

### SWMLService Class

Base class providing SWML document generation and HTTP service capabilities. `AgentBase` extends this class.

#### Key Methods

##### `get_document() -> Dict[str, Any]`
Generate the complete SWML document for the service.

##### `handle_request(method: str, url: str, headers: Dict[str, str], body: Optional[Dict[str, Any]] = None) -> Tuple[int, Dict[str, str], str]`
Framework-free request dispatch: runs proxy detection, basic auth, routing callbacks, and `on_request`, and returns a `(status_code, headers, body)` tuple.

### Dynamic Configuration

The dynamic configuration callback receives a copy of the agent made for the request, and configures it based on the request's data. Its changes apply to that request only.

**Usage:**
```python
def dynamic_config(query_params, body_params, headers, agent):
    # Configure based on request
    if query_params.get("lang") == "es":
        agent.add_language("Spanish", "es-ES", "nova.luna")
    
    # Customer-specific configuration
    customer_id = headers.get("X-Customer-ID")
    if customer_id:
        agent.set_global_data({"customer_id": customer_id})
        agent.prompt_add_section("Customer Context", f"You are helping customer {customer_id}")
    
    # Add skills dynamically
    if query_params.get("enable_search") == "true":
        agent.add_skill("web_search", {"provider": "google"})

agent.set_dynamic_config_callback(dynamic_config)
```

---

## Environment Variables

The SDK supports various environment variables for configuration:

### Authentication

These variables set the agent's basic auth credentials:

- `SWML_BASIC_AUTH_USER`: Basic auth username
- `SWML_BASIC_AUTH_PASSWORD`: Basic auth password

### SSL/HTTPS

These variables configure HTTPS:

- `SWML_SSL_ENABLED`: Enable SSL (true/false)
- `SWML_SSL_CERT_PATH`: Path to SSL certificate
- `SWML_SSL_KEY_PATH`: Path to SSL private key
- `SWML_DOMAIN`: Domain name for SSL

### Proxy Support

This variable overrides the URL the SDK generates behind a reverse proxy:

- `SWML_PROXY_URL_BASE`: Base URL for proxy server

### Skills Configuration

The `web_search` skill falls back to these variables when `api_key` or `search_engine_id` is omitted:

- `GOOGLE_SEARCH_API_KEY`: Google Custom Search API key
- `GOOGLE_SEARCH_ENGINE_ID`: Google Custom Search Engine ID

### Usage

An agent can also set these variables itself before reading its configuration:

<!-- snippet: no-run skill setup needs an external backend/API key (e.g. web_search, mcp_gateway) -->
```python
import os

# Set environment variables
os.environ["SWML_BASIC_AUTH_USER"] = "admin"
os.environ["SWML_BASIC_AUTH_PASSWORD"] = "secret"
os.environ["GOOGLE_SEARCH_API_KEY"] = "your-api-key"

# Agent will automatically use these
agent = AgentBase("My Agent")
agent.add_skill("web_search", {
    "search_engine_id": "your-engine-id"
    # api_key will be read from environment
})
```

---

## Complete Example

Here's a comprehensive example using multiple SDK components:

<!-- snippet: no-run starts a blocking server/client (covered by SNIPPET-COMPILE + EXAMPLES-RUN) -->
```python
from signalwire import AgentBase, FunctionResult, DataMap

class ComprehensiveAgent(AgentBase):
    def __init__(self):
        super().__init__(
            name="Comprehensive Agent",
            auto_answer=True,
            record_call=True
        )
        
        # Configure voice and language
        self.add_language("English", "en-US", "rime.spore",
                         speech_fillers=["Let me check...", "One moment..."])
        
        # Add speech recognition hints
        self.add_hints(["SignalWire", "customer service", "technical support"])
        
        # Configure AI parameters
        self.set_params({
            "ai_model": "gpt-4.1-nano",
            "end_of_speech_timeout": 800,
            "temperature": 0.7
        })
        
        # Add skills
        self.add_skill("datetime")
        self.add_skill("math")
        self.add_skill("web_search", {
            "api_key": "your-google-api-key",
            "search_engine_id": "your-engine-id",
            "num_results": 3
        })
        
        # Set up structured workflow
        self._setup_contexts()
        
        # Add custom tools
        self._register_custom_tools()
        
        # Set global data
        self.set_global_data({
            "company_name": "Acme Corp",
            "support_hours": "9 AM - 5 PM EST",
            "version": "2.0"
        })
    
    def _setup_contexts(self):
        """Set up structured workflow contexts"""
        contexts = self.define_contexts()
        
        # Greeting context
        greeting = contexts.add_context("greeting")
        greeting.add_step("welcome") \
            .set_text("Hello! Welcome to Acme Corp support. How can I help you today?") \
            .set_step_criteria("Customer has explained their issue") \
            .set_valid_steps(["next"])
        
        greeting.add_step("categorize") \
            .add_section("Current Task", "Categorize the customer's request") \
            .add_bullets("Categories", [
                "Technical issue - use diagnostic tools",
                "Billing question - transfer to billing",
                "General inquiry - handle directly"
            ]) \
            .set_functions(["transfer_to_billing", "run_diagnostics"]) \
            .set_step_criteria("Request categorized and action taken")
        
        # Technical support context
        tech = contexts.add_context("technical_support")
        tech.add_step("diagnose") \
            .set_text("Let me run some diagnostics to identify the issue.") \
            .set_functions(["run_diagnostics", "check_system_status"]) \
            .set_step_criteria("Diagnostics completed") \
            .set_valid_steps(["resolve"])
        
        tech.add_step("resolve") \
            .set_text("Based on the diagnostics, here's how we'll fix this.") \
            .set_functions(["apply_fix", "schedule_technician"]) \
            .set_step_criteria("Issue resolved or escalated")
    
    def _register_custom_tools(self):
        """Register custom DataMap tools"""
        
        # Customer lookup tool
        lookup_tool = (DataMap("lookup_customer")
            .description("Look up customer information")
            .parameter("customer_id", "string", "Customer ID", required=True)
            .webhook("GET", "https://api.company.com/customers/${args.customer_id}",
                    headers={"Authorization": "Bearer YOUR_TOKEN"})
            .output(FunctionResult("Customer: ${name}, Status: ${status}"))
            .error_keys(["error"])
        )
        
        self.register_swaig_function(lookup_tool.to_swaig_function())
        
        # System control tool
        control_tool = (DataMap("system_control")
            .description("Control system functions")
            .parameter("action", "string", "Action to perform", required=True)
            .parameter("target", "string", "Target system")
            .expression("${args.action}", r"restart|reboot",
                       FunctionResult("Restarting ${args.target}")
                       .add_action("restart_system", {"target": "${args.target}"}))
            .expression("${args.action}", r"status|check",
                       FunctionResult("Checking ${args.target} status")
                       .add_action("check_status", {"target": "${args.target}"}))
        )
        
        self.register_swaig_function(control_tool.to_swaig_function())
    
    @AgentBase.tool(
        description="Transfer call to billing department",
        parameters={"type": "object", "properties": {}}
    )
    def transfer_to_billing(self, args, raw_data):
        """Transfer to billing with state tracking"""
        return (FunctionResult("Transferring you to our billing department")
                .update_global_data({"last_action": "transfer_to_billing"})
                .connect("billing@company.com", final=False))
    
    def on_summary(self, summary, raw_data):
        """Handle conversation summaries"""
        print(f"Conversation completed: {summary}")
        # Could save to database, send notifications, etc.

# Run the agent
if __name__ == "__main__":
    agent = ComprehensiveAgent()
    agent.run()
```

This concludes the complete API reference for the SignalWire SDK. The SDK provides a comprehensive framework for building sophisticated AI agents with modular capabilities, structured workflows, persistent state, and deployment across multiple environments.