From 0cc266858fc71138339ae232760600f6dd71870d Mon Sep 17 00:00:00 2001 From: Iftekhairul Alam Date: Sat, 18 Jul 2026 14:25:18 +0600 Subject: [PATCH 1/3] Build standard NextGenSwitch Python SDK --- .github/workflows/tests.yml | 28 +++++ .gitignore | 11 ++ LICENSE | 21 ++++ README.md | 207 +++++++++++++++++++++++++++++++- examples/create_call.py | 26 ++++ examples/modify_call.py | 19 +++ examples/record_call.py | 26 ++++ examples/stream_to_ai.py | 19 +++ pyproject.toml | 57 +++++++++ src/nextgenswitch/__init__.py | 23 ++++ src/nextgenswitch/client.py | 181 ++++++++++++++++++++++++++++ src/nextgenswitch/exceptions.py | 22 ++++ src/nextgenswitch/py.typed | 1 + src/nextgenswitch/responses.py | 15 +++ src/nextgenswitch/voice.py | 127 ++++++++++++++++++++ src/nextgenswitch/webhooks.py | 51 ++++++++ tests/test_async_client.py | 22 ++++ tests/test_client.py | 66 ++++++++++ tests/test_voice.py | 41 +++++++ tests/test_webhooks.py | 18 +++ 20 files changed, 980 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/tests.yml create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 examples/create_call.py create mode 100644 examples/modify_call.py create mode 100644 examples/record_call.py create mode 100644 examples/stream_to_ai.py create mode 100644 pyproject.toml create mode 100644 src/nextgenswitch/__init__.py create mode 100644 src/nextgenswitch/client.py create mode 100644 src/nextgenswitch/exceptions.py create mode 100644 src/nextgenswitch/py.typed create mode 100644 src/nextgenswitch/responses.py create mode 100644 src/nextgenswitch/voice.py create mode 100644 src/nextgenswitch/webhooks.py create mode 100644 tests/test_async_client.py create mode 100644 tests/test_client.py create mode 100644 tests/test_voice.py create mode 100644 tests/test_webhooks.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..dba3eee --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,28 @@ +name: Tests + +on: + push: + pull_request: + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python: ["3.9", "3.10", "3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + cache: pip + - run: python -m pip install --upgrade pip + - run: python -m pip install -e ".[dev]" + - run: ruff check . + - run: mypy src/nextgenswitch + - run: pytest + - run: python -m build + - run: python -m twine check dist/* diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..94dc61a --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +.coverage +.mypy_cache/ +.pytest_cache/ +.ruff_cache/ +.venv/ +build/ +dist/ +.env diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..af2038b --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Infosoftbd Solutions + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index e53521c..15e9a10 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,208 @@ # NextGenSwitch Python SDK -SDK development is tracked in pull requests. +[![Tests](https://github.com/nextgenswitch/nextgenswitch-python/actions/workflows/tests.yml/badge.svg)](https://github.com/nextgenswitch/nextgenswitch-python/actions/workflows/tests.yml) +[![Python](https://img.shields.io/badge/Python-3.9%2B-3776AB.svg)](https://www.python.org/) +[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE) + +The official Python SDK for the [NextGenSwitch Programmable Voice API](https://nextgenswitch.com/docs/programmable-voice-api/). Create and modify calls with synchronous or asynchronous clients, build escaped Voice XML, stream audio to AI services, and parse Gather and Dial callbacks. + +## Requirements + +- Python 3.9 or newer +- A NextGenSwitch deployment and API credentials + +## Installation + +```bash +pip install nextgenswitch +``` + +Until the first PyPI release, install from GitHub: + +```bash +pip install "nextgenswitch @ git+https://github.com/nextgenswitch/nextgenswitch-python.git" +``` + +## Configure the Client + +```python +import os +from nextgenswitch import Client + +client = Client( + os.environ["NEXTGENSWITCH_BASE_URL"], + os.environ["NEXTGENSWITCH_AUTHORIZATION"], + os.environ["NEXTGENSWITCH_AUTHORIZATION_SECRET"], +) +``` + +The SDK sends the documented `X-Authorization` and `X-Authorization-Secret` headers. Keep these values in environment variables or a secret manager and use HTTPS for remote deployments. + +## Create a Call + +```python +from nextgenswitch import VoiceResponse + +flow = VoiceResponse().say("Welcome to NextGenSwitch.").gather( + action="https://example.com/gather", + method="POST", + numDigits=1, + timeout=10, + children=lambda gather: gather.say("Press one for sales."), +) + +result = client.create_call( + "2001", + "1001", + response_xml=flow, + status_callback="https://example.com/call-status", +) +print(result.data) +``` + +Use `response_url="https://example.com/call-flow.xml"` instead of `response_xml` to provide a hosted XML document. Exactly one response source is required. + +## Modify an Active Call + +```python +updated = ( + VoiceResponse() + .pause(2) + .say("Your call flow has been updated.") + .dial("1000", answerOnBridge=True) +) + +client.modify_call("CALL-123", updated) +``` + +## Async Client + +```python +import asyncio +import os + +from nextgenswitch import AsyncClient, VoiceResponse + +async def main() -> None: + async with AsyncClient( + os.environ["NEXTGENSWITCH_BASE_URL"], + os.environ["NEXTGENSWITCH_AUTHORIZATION"], + os.environ["NEXTGENSWITCH_AUTHORIZATION_SECRET"], + ) as client: + result = await client.create_call( + "2001", + "1001", + response_xml=VoiceResponse().say("Hello from async Python."), + ) + print(result.data) + +asyncio.run(main()) +``` + +## Voice XML + +| XML verb | Python method | +| --- | --- | +| `` | `say(text, **attributes)` | +| `` | `play(url, **attributes)` | +| `` | `gather(children=..., **attributes)` | +| `` | `dial(to, children=..., **attributes)` | +| `` | `record(**attributes)` | +| `` | `stream(url, parameters=..., **attributes)` | +| `` | `hangup()` | +| `` | `pause(seconds)` | +| `` | `redirect(url, method=...)` | +| `` | `bridge(call_id, bridge_after_establish=...)` | +| `` | `leave()` | + +Python's XML library escapes text and attributes instead of concatenating untrusted XML strings. + +## Record a Call + +```python +flow = ( + VoiceResponse() + .say("Please leave your message after the beep.") + .record( + action="https://example.com/recording", + method="POST", + timeout=5, + finishOnKey="#", + transcribe=True, + trim=True, + beep=True, + ) + .hangup() +) +client.create_call("2001", "1001", response_xml=flow) +``` + +## Stream to an AI Voice Service + +```python +flow = VoiceResponse().stream( + "wss://voice.example.com/session", + parameters={"session_id": "session-123", "tenant": "example"}, + name="assistant-stream", +) +``` + +Resolve AI-provider secrets on the WebSocket service. Never put provider API keys in Voice XML. + +## Parse Webhooks + +```python +from nextgenswitch import GatherResult + +gather = GatherResult.from_mapping(request.form) +if gather.digits == "1": + response = VoiceResponse().say("Connecting sales.").dial("1001") +else: + response = VoiceResponse().say("No valid selection received.").hangup() +``` + +Use `DialResult.from_mapping(payload)` for documented Dial action fields. Callback signature verification is not documented by the current API; restrict endpoints, require TLS, validate expected fields, and add deployment-appropriate authentication. + +## Errors + +- `ValidationError`: invalid SDK input +- `ApiError`: non-2xx response, with `status_code` and `response_body` +- `TransportError`: network or HTTP transport failure +- `NextGenSwitchError`: base SDK exception + +## Examples + +Configure `NEXTGENSWITCH_BASE_URL`, `NEXTGENSWITCH_AUTHORIZATION`, and `NEXTGENSWITCH_AUTHORIZATION_SECRET`, then adapt: + +- [Create call](examples/create_call.py) +- [Modify active call](examples/modify_call.py) +- [Record caller audio](examples/record_call.py) +- [Stream to AI](examples/stream_to_ai.py) + +All `example.com` URLs are placeholders for TLS endpoints you control. + +## Development + +```bash +python -m venv .venv +source .venv/bin/activate +python -m pip install -e ".[dev]" +ruff check . +mypy src/nextgenswitch +pytest +python -m build +python -m twine check dist/* +``` + +GitHub Actions validates Python 3.9–3.13. + +## Documentation + +- [Programmable Voice API](https://nextgenswitch.com/docs/programmable-voice-api/) +- [NextGenSwitch documentation](https://nextgenswitch.com/docs/) +- [NextGenSwitch website](https://nextgenswitch.com/) +- [Contact NextGenSwitch](https://nextgenswitch.com/contact/) + +## License + +[MIT](LICENSE) diff --git a/examples/create_call.py b/examples/create_call.py new file mode 100644 index 0000000..4ae17cf --- /dev/null +++ b/examples/create_call.py @@ -0,0 +1,26 @@ +import os + +from nextgenswitch import Client, VoiceResponse + +client = Client( + os.environ["NEXTGENSWITCH_BASE_URL"], + os.environ["NEXTGENSWITCH_AUTHORIZATION"], + os.environ["NEXTGENSWITCH_AUTHORIZATION_SECRET"], +) + +flow = VoiceResponse().say("Welcome to NextGenSwitch.").gather( + action="https://example.com/gather", + method="POST", + numDigits=1, + timeout=10, + children=lambda gather: gather.say("Press one for sales or two for support."), +) + +result = client.create_call( + "2001", + "1001", + response_xml=flow, + status_callback="https://example.com/call-status", +) +print(result.data) +client.close() diff --git a/examples/modify_call.py b/examples/modify_call.py new file mode 100644 index 0000000..cf21988 --- /dev/null +++ b/examples/modify_call.py @@ -0,0 +1,19 @@ +import os +import sys + +from nextgenswitch import Client, VoiceResponse + +if len(sys.argv) != 2: + raise SystemExit("Usage: python examples/modify_call.py CALL_ID") + +client = Client( + os.environ["NEXTGENSWITCH_BASE_URL"], + os.environ["NEXTGENSWITCH_AUTHORIZATION"], + os.environ["NEXTGENSWITCH_AUTHORIZATION_SECRET"], +) +flow = VoiceResponse().pause(1).say("Your call flow has been updated.").dial( + "1000", answerOnBridge=True, timeLimit=300 +) +result = client.modify_call(sys.argv[1], flow) +print(result.data) +client.close() diff --git a/examples/record_call.py b/examples/record_call.py new file mode 100644 index 0000000..c610f62 --- /dev/null +++ b/examples/record_call.py @@ -0,0 +1,26 @@ +import os + +from nextgenswitch import Client, VoiceResponse + +client = Client( + os.environ["NEXTGENSWITCH_BASE_URL"], + os.environ["NEXTGENSWITCH_AUTHORIZATION"], + os.environ["NEXTGENSWITCH_AUTHORIZATION_SECRET"], +) +flow = ( + VoiceResponse() + .say("Please leave your message after the beep.") + .record( + action="https://example.com/recording", + method="POST", + timeout=5, + finishOnKey="#", + transcribe=True, + trim=True, + beep=True, + ) + .say("Thank you. Goodbye.") + .hangup() +) +print(client.create_call("2001", "1001", response_xml=flow).data) +client.close() diff --git a/examples/stream_to_ai.py b/examples/stream_to_ai.py new file mode 100644 index 0000000..eaedcf2 --- /dev/null +++ b/examples/stream_to_ai.py @@ -0,0 +1,19 @@ +import os +import secrets + +from nextgenswitch import Client, VoiceResponse + +client = Client( + os.environ["NEXTGENSWITCH_BASE_URL"], + os.environ["NEXTGENSWITCH_AUTHORIZATION"], + os.environ["NEXTGENSWITCH_AUTHORIZATION_SECRET"], +) +flow = VoiceResponse().say("Connecting the virtual assistant.").stream( + "wss://voice.example.com/session", + parameters={"session_id": secrets.token_hex(16), "tenant": "example"}, + name="ai-assistant", +) +print(client.create_call("2001", "1001", response_xml=flow).data) +client.close() + +# Resolve AI-provider credentials on the WebSocket service, not in Voice XML. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..5896d81 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,57 @@ +[build-system] +requires = ["setuptools>=69"] +build-backend = "setuptools.build_meta" + +[project] +name = "nextgenswitch" +version = "0.1.0" +description = "Official Python SDK for the NextGenSwitch Programmable Voice API and Voice XML." +readme = "README.md" +requires-python = ">=3.9" +license = { text = "MIT" } +authors = [{ name = "Infosoftbd Solutions" }] +keywords = ["nextgenswitch", "programmable-voice", "voice-api", "sip", "voip", "pbx", "contact-center"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Typing :: Typed", +] +dependencies = ["httpx>=0.27,<1"] + +[project.optional-dependencies] +dev = ["build>=1.2", "mypy>=1.10", "pytest>=8.0", "ruff>=0.5", "twine>=5.1"] + +[project.urls] +Homepage = "https://nextgenswitch.com/" +Documentation = "https://nextgenswitch.com/docs/programmable-voice-api/" +Repository = "https://github.com/nextgenswitch/nextgenswitch-python" +Issues = "https://github.com/nextgenswitch/nextgenswitch-python/issues" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.setuptools.package-data] +nextgenswitch = ["py.typed"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-q" + +[tool.ruff] +target-version = "py39" +line-length = 100 + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B"] + +[tool.mypy] +python_version = "3.9" +strict = true +packages = ["nextgenswitch"] diff --git a/src/nextgenswitch/__init__.py b/src/nextgenswitch/__init__.py new file mode 100644 index 0000000..2b13eaf --- /dev/null +++ b/src/nextgenswitch/__init__.py @@ -0,0 +1,23 @@ +"""NextGenSwitch Python SDK.""" + +from .client import AsyncClient, Client +from .exceptions import ApiError, NextGenSwitchError, TransportError, ValidationError +from .responses import ApiResponse +from .voice import VoiceNode, VoiceResponse +from .webhooks import DialResult, GatherResult + +__all__ = [ + "ApiError", + "ApiResponse", + "AsyncClient", + "Client", + "DialResult", + "GatherResult", + "NextGenSwitchError", + "TransportError", + "ValidationError", + "VoiceNode", + "VoiceResponse", +] + +__version__ = "0.1.0" diff --git a/src/nextgenswitch/client.py b/src/nextgenswitch/client.py new file mode 100644 index 0000000..77ff6af --- /dev/null +++ b/src/nextgenswitch/client.py @@ -0,0 +1,181 @@ +"""Synchronous and asynchronous clients for the NextGenSwitch Voice API.""" + +from __future__ import annotations + +from typing import Any, Optional, Union +from urllib.parse import quote + +import httpx + +from .exceptions import ApiError, TransportError, ValidationError +from .responses import ApiResponse +from .voice import VoiceResponse + +VoiceXml = Union[VoiceResponse, str] + + +def _validate_configuration(base_url: str, authorization: str, secret: str) -> str: + normalized = base_url.rstrip("/") + if not normalized or not authorization or not secret: + raise ValidationError("Base URL and both authorization values are required.") + return normalized + + +def _call_payload( + to: str, + from_: str, + response_xml: Optional[VoiceXml], + response_url: Optional[str], + status_callback: Optional[str], +) -> dict[str, str]: + if not to or not from_: + raise ValidationError("The to and from_ values are required.") + if (response_xml is None) == (response_url is None): + raise ValidationError("Provide exactly one of response_xml or response_url.") + + payload = {"to": to, "from": from_} + if response_xml is not None: + payload["responseXml"] = str(response_xml) + if response_url is not None: + payload["response"] = response_url + if status_callback is not None: + payload["statusCallback"] = status_callback + return payload + + +def _api_response(response: httpx.Response) -> ApiResponse: + try: + decoded: Any = response.json() + except ValueError: + decoded = None + data = decoded if isinstance(decoded, dict) else None + if not response.is_success: + message = str(data.get("message")) if data and "message" in data else ( + f"NextGenSwitch API returned HTTP {response.status_code}." + ) + raise ApiError(message, response.status_code, response.text) + return ApiResponse(response.status_code, data, response.text) + + +class Client: + """Blocking NextGenSwitch API client.""" + + def __init__( + self, + base_url: str, + authorization: str, + authorization_secret: str, + *, + timeout: float = 30.0, + http_client: Optional[httpx.Client] = None, + ) -> None: + self.base_url = _validate_configuration(base_url, authorization, authorization_secret) + self._owns_client = http_client is None + self._client = http_client or httpx.Client(timeout=timeout) + self._headers = { + "Accept": "application/json", + "X-Authorization": authorization, + "X-Authorization-Secret": authorization_secret, + } + + def create_call( + self, + to: str, + from_: str, + *, + response_xml: Optional[VoiceXml] = None, + response_url: Optional[str] = None, + status_callback: Optional[str] = None, + ) -> ApiResponse: + payload = _call_payload(to, from_, response_xml, response_url, status_callback) + return self._request("POST", "/api/v1/call", data=payload) + + def modify_call(self, call_id: str, response_xml: VoiceXml) -> ApiResponse: + if not call_id: + raise ValidationError("Call ID is required.") + return self._request( + "PUT", + f"/api/v1/call/{quote(call_id, safe='')}", + json={"responseXml": str(response_xml)}, + ) + + def _request(self, method: str, path: str, **kwargs: Any) -> ApiResponse: + try: + response = self._client.request( + method, self.base_url + path, headers=self._headers, **kwargs + ) + except httpx.HTTPError as exc: + raise TransportError(f"NextGenSwitch request failed: {exc}") from exc + return _api_response(response) + + def close(self) -> None: + if self._owns_client: + self._client.close() + + def __enter__(self) -> Client: + return self + + def __exit__(self, *_: object) -> None: + self.close() + + +class AsyncClient: + """Async NextGenSwitch API client.""" + + def __init__( + self, + base_url: str, + authorization: str, + authorization_secret: str, + *, + timeout: float = 30.0, + http_client: Optional[httpx.AsyncClient] = None, + ) -> None: + self.base_url = _validate_configuration(base_url, authorization, authorization_secret) + self._owns_client = http_client is None + self._client = http_client or httpx.AsyncClient(timeout=timeout) + self._headers = { + "Accept": "application/json", + "X-Authorization": authorization, + "X-Authorization-Secret": authorization_secret, + } + + async def create_call( + self, + to: str, + from_: str, + *, + response_xml: Optional[VoiceXml] = None, + response_url: Optional[str] = None, + status_callback: Optional[str] = None, + ) -> ApiResponse: + payload = _call_payload(to, from_, response_xml, response_url, status_callback) + return await self._request("POST", "/api/v1/call", data=payload) + + async def modify_call(self, call_id: str, response_xml: VoiceXml) -> ApiResponse: + if not call_id: + raise ValidationError("Call ID is required.") + return await self._request( + "PUT", + f"/api/v1/call/{quote(call_id, safe='')}", + json={"responseXml": str(response_xml)}, + ) + + async def _request(self, method: str, path: str, **kwargs: Any) -> ApiResponse: + try: + response = await self._client.request( + method, self.base_url + path, headers=self._headers, **kwargs + ) + except httpx.HTTPError as exc: + raise TransportError(f"NextGenSwitch request failed: {exc}") from exc + return _api_response(response) + + async def aclose(self) -> None: + if self._owns_client: + await self._client.aclose() + + async def __aenter__(self) -> AsyncClient: + return self + + async def __aexit__(self, *_: object) -> None: + await self.aclose() diff --git a/src/nextgenswitch/exceptions.py b/src/nextgenswitch/exceptions.py new file mode 100644 index 0000000..e81d965 --- /dev/null +++ b/src/nextgenswitch/exceptions.py @@ -0,0 +1,22 @@ +"""Exception hierarchy for the NextGenSwitch SDK.""" + + +class NextGenSwitchError(Exception): + """Base SDK exception.""" + + +class ValidationError(NextGenSwitchError, ValueError): + """Raised when SDK input is invalid.""" + + +class TransportError(NextGenSwitchError): + """Raised when the API cannot be reached.""" + + +class ApiError(NextGenSwitchError): + """Raised for non-successful API responses.""" + + def __init__(self, message: str, status_code: int, response_body: str = "") -> None: + super().__init__(message) + self.status_code = status_code + self.response_body = response_body diff --git a/src/nextgenswitch/py.typed b/src/nextgenswitch/py.typed new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/nextgenswitch/py.typed @@ -0,0 +1 @@ + diff --git a/src/nextgenswitch/responses.py b/src/nextgenswitch/responses.py new file mode 100644 index 0000000..5c65dc0 --- /dev/null +++ b/src/nextgenswitch/responses.py @@ -0,0 +1,15 @@ +"""Typed API responses.""" + +from dataclasses import dataclass +from typing import Any, Mapping, Optional + + +@dataclass(frozen=True) +class ApiResponse: + status_code: int + data: Optional[Mapping[str, Any]] + body: str + + @property + def successful(self) -> bool: + return 200 <= self.status_code < 300 diff --git a/src/nextgenswitch/voice.py b/src/nextgenswitch/voice.py new file mode 100644 index 0000000..41deb59 --- /dev/null +++ b/src/nextgenswitch/voice.py @@ -0,0 +1,127 @@ +"""Safe builder for NextGenSwitch Voice XML.""" + +from __future__ import annotations + +from typing import Any, Callable, Mapping, Optional +from xml.etree import ElementTree as ET + + +def _attributes(values: Mapping[str, Any]) -> dict[str, str]: + normalized: dict[str, str] = {} + for name, value in values.items(): + if value is None: + continue + if isinstance(value, bool): + normalized[name] = "true" if value else "false" + else: + normalized[name] = str(value) + return normalized + + +class VoiceNode: + """A restricted nested node used inside Gather and Dial.""" + + def __init__(self, element: ET.Element) -> None: + self._element = element + + def say(self, text: str, **attributes: Any) -> VoiceNode: + child = ET.SubElement(self._element, "Say", _attributes(attributes)) + child.text = text + return self + + def play(self, url: str, **attributes: Any) -> VoiceNode: + child = ET.SubElement(self._element, "Play", _attributes(attributes)) + child.text = url + return self + + +class VoiceResponse: + """Fluent, XML-escaping builder for documented NextGenSwitch voice verbs.""" + + def __init__(self) -> None: + self._root = ET.Element("Response") + + def say(self, text: str, **attributes: Any) -> VoiceResponse: + child = ET.SubElement(self._root, "Say", _attributes(attributes)) + child.text = text + return self + + def play(self, url: str, **attributes: Any) -> VoiceResponse: + child = ET.SubElement(self._root, "Play", _attributes(attributes)) + child.text = url + return self + + def gather( + self, + *, + children: Optional[Callable[[VoiceNode], None]] = None, + **attributes: Any, + ) -> VoiceResponse: + element = ET.SubElement(self._root, "Gather", _attributes(attributes)) + if children is not None: + children(VoiceNode(element)) + return self + + def dial( + self, + to: str, + *, + children: Optional[Callable[[VoiceNode], None]] = None, + **attributes: Any, + ) -> VoiceResponse: + element = ET.SubElement(self._root, "Dial", _attributes({"to": to, **attributes})) + if children is not None: + children(VoiceNode(element)) + return self + + def record(self, **attributes: Any) -> VoiceResponse: + ET.SubElement(self._root, "Record", _attributes(attributes)) + return self + + def stream( + self, + url: str, + *, + parameters: Optional[Mapping[str, Any]] = None, + **attributes: Any, + ) -> VoiceResponse: + connect = ET.SubElement(self._root, "Connect") + stream = ET.SubElement(connect, "Stream", _attributes({"url": url, **attributes})) + for name, value in (parameters or {}).items(): + ET.SubElement(stream, "Parameter", _attributes({"name": name, "value": value})) + return self + + def hangup(self) -> VoiceResponse: + ET.SubElement(self._root, "Hangup") + return self + + def pause(self, length: int) -> VoiceResponse: + if length < 0: + raise ValueError("Pause length cannot be negative.") + ET.SubElement(self._root, "Pause", {"length": str(length)}) + return self + + def redirect(self, url: str, *, method: str = "POST") -> VoiceResponse: + element = ET.SubElement(self._root, "Redirect", {"method": method.upper()}) + element.text = url + return self + + def bridge(self, call_id: str, *, bridge_after_establish: bool = True) -> VoiceResponse: + element = ET.SubElement( + self._root, + "Bridge", + {"bridgeAfterEstablish": "true" if bridge_after_establish else "false"}, + ) + element.text = call_id + return self + + def leave(self) -> VoiceResponse: + ET.SubElement(self._root, "Leave") + return self + + def to_xml(self) -> str: + body = ET.tostring(self._root, encoding="unicode", short_empty_elements=True) + return '\n' + body + + def __str__(self) -> str: + return self.to_xml() diff --git a/src/nextgenswitch/webhooks.py b/src/nextgenswitch/webhooks.py new file mode 100644 index 0000000..410d12f --- /dev/null +++ b/src/nextgenswitch/webhooks.py @@ -0,0 +1,51 @@ +"""Typed callback payloads from Gather and Dial actions.""" + +from dataclasses import dataclass +from typing import Any, Mapping, Optional + + +@dataclass(frozen=True) +class GatherResult: + call_id: str + digits: Optional[str] = None + speech_result: Optional[str] = None + confidence: Optional[float] = None + voice: Optional[str] = None + from_: Optional[str] = None + to: Optional[str] = None + + @classmethod + def from_mapping(cls, payload: Mapping[str, Any]) -> "GatherResult": + confidence = payload.get("confidence") + return cls( + call_id=str(payload.get("call_id", "")), + digits=str(payload["digits"]) if "digits" in payload else None, + speech_result=str(payload["speech_result"]) if "speech_result" in payload else None, + confidence=float(confidence) if confidence not in (None, "") else None, + voice=str(payload["voice"]) if "voice" in payload else None, + from_=str(payload["event_from"]) if "event_from" in payload else None, + to=str(payload["event_to"]) if "event_to" in payload else None, + ) + + +@dataclass(frozen=True) +class DialResult: + call_id: str + bridge_call_id: Optional[str] + established: bool + duration: int + waiting_duration: int + record_file: Optional[str] + + @classmethod + def from_mapping(cls, payload: Mapping[str, Any]) -> "DialResult": + return cls( + call_id=str(payload.get("call_id", "")), + bridge_call_id=( + str(payload["bridge_call_id"]) if payload.get("bridge_call_id") else None + ), + established=int(payload.get("dial_status", 0)) == 1, + duration=int(payload.get("duration", 0)), + waiting_duration=int(payload.get("waiting_duration", 0)), + record_file=str(payload["record_file"]) if payload.get("record_file") else None, + ) diff --git a/tests/test_async_client.py b/tests/test_async_client.py new file mode 100644 index 0000000..330c3e0 --- /dev/null +++ b/tests/test_async_client.py @@ -0,0 +1,22 @@ +import asyncio + +import httpx + +from nextgenswitch import AsyncClient, VoiceResponse + + +def test_async_create_call() -> None: + async def run() -> None: + async def handler(request: httpx.Request) -> httpx.Response: + assert request.headers["X-Authorization-Secret"] == "secret" + return httpx.Response(201, json={"call_id": "CALL-ASYNC"}) + + http = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + client = AsyncClient("https://switch.example.com", "code", "secret", http_client=http) + result = await client.create_call( + "2001", "1001", response_xml=VoiceResponse().say("Hello") + ) + assert result.data == {"call_id": "CALL-ASYNC"} + await http.aclose() + + asyncio.run(run()) diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..7b9282f --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,66 @@ +import json + +import httpx +import pytest + +from nextgenswitch import ApiError, Client, ValidationError, VoiceResponse + + +def test_creates_call_with_authentication_and_xml() -> None: + captured: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(request) + return httpx.Response(201, json={"call_id": "CALL-123"}) + + http = httpx.Client(transport=httpx.MockTransport(handler)) + client = Client("https://switch.example.com/", "code", "secret", http_client=http) + result = client.create_call("2001", "1001", response_xml=VoiceResponse().say("Hello")) + + assert result.status_code == 201 + assert result.data == {"call_id": "CALL-123"} + assert captured[0].headers["X-Authorization"] == "code" + assert "responseXml=" in captured[0].content.decode() + + +def test_modifies_encoded_call_id() -> None: + captured: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured.append(request) + return httpx.Response(200, json={"updated": True}) + + client = Client( + "https://switch.example.com", + "code", + "secret", + http_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + client.modify_call("CALL/123", VoiceResponse().hangup()) + + assert captured[0].url.path == "/api/v1/call/CALL/123" + assert json.loads(captured[0].content)["responseXml"].startswith(" None: + client = Client("https://switch.example.com", "code", "secret") + with pytest.raises(ValidationError): + client.create_call("2", "1") + client.close() + + +def test_raises_typed_api_error() -> None: + def handler(_: httpx.Request) -> httpx.Response: + return httpx.Response(401, json={"message": "Unauthorized"}) + + client = Client( + "https://switch.example.com", + "code", + "secret", + http_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + with pytest.raises(ApiError) as caught: + client.modify_call("CALL-123", VoiceResponse().hangup()) + + assert caught.value.status_code == 401 + assert str(caught.value) == "Unauthorized" diff --git a/tests/test_voice.py b/tests/test_voice.py new file mode 100644 index 0000000..1f0e058 --- /dev/null +++ b/tests/test_voice.py @@ -0,0 +1,41 @@ +from xml.etree import ElementTree as ET + +import pytest + +from nextgenswitch import VoiceResponse + + +def test_builds_escaped_voice_xml() -> None: + response = ( + VoiceResponse() + .say("Sales & support", loop=2) + .gather( + action="https://example.com/input", + input="dtmf speech", + numDigits=1, + children=lambda node: node.say("Press "), + ) + .dial("+15551234567", answerOnBridge=True) + .pause(2) + .hangup() + ) + + root = ET.fromstring(response.to_xml()) + assert root.tag == "Response" + assert "Sales & support" in response.to_xml() + assert 'answerOnBridge="true"' in response.to_xml() + + +def test_builds_stream_parameters() -> None: + xml = VoiceResponse().stream( + "wss://voice.example.com/ws", parameters={"session": "abc"}, name="assistant" + ).to_xml() + + assert "" in xml + assert 'url="wss://voice.example.com/ws"' in xml + assert 'name="session" value="abc"' in xml + + +def test_rejects_negative_pause() -> None: + with pytest.raises(ValueError): + VoiceResponse().pause(-1) diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py new file mode 100644 index 0000000..47ff9c6 --- /dev/null +++ b/tests/test_webhooks.py @@ -0,0 +1,18 @@ +from nextgenswitch import DialResult, GatherResult + + +def test_parses_gather_callback() -> None: + result = GatherResult.from_mapping( + {"call_id": "C1", "digits": "1234", "confidence": "0.95"} + ) + assert result.call_id == "C1" + assert result.digits == "1234" + assert result.confidence == 0.95 + + +def test_parses_dial_callback() -> None: + result = DialResult.from_mapping( + {"call_id": "C1", "dial_status": "1", "duration": "84"} + ) + assert result.established is True + assert result.duration == 84 From dfcc6912ac9a6d1d7b9d91a2d004564d2c2e7cf4 Mon Sep 17 00:00:00 2001 From: Iftekhairul Alam Date: Sat, 18 Jul 2026 14:30:55 +0600 Subject: [PATCH 2/3] Align Ruff rules with Python 3.9 support --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 5896d81..4de46d2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,7 +49,7 @@ target-version = "py39" line-length = 100 [tool.ruff.lint] -select = ["E", "F", "I", "UP", "B"] +select = ["E", "F", "I", "B"] [tool.mypy] python_version = "3.9" From 9c077216b9e75b7d56a53aa174eecbb193edece6 Mon Sep 17 00:00:00 2001 From: Iftekhairul Alam Date: Sat, 18 Jul 2026 14:32:01 +0600 Subject: [PATCH 3/3] Fix Python 3.9 callback type narrowing --- src/nextgenswitch/webhooks.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/nextgenswitch/webhooks.py b/src/nextgenswitch/webhooks.py index 410d12f..cf57a09 100644 --- a/src/nextgenswitch/webhooks.py +++ b/src/nextgenswitch/webhooks.py @@ -17,11 +17,14 @@ class GatherResult: @classmethod def from_mapping(cls, payload: Mapping[str, Any]) -> "GatherResult": confidence = payload.get("confidence") + confidence_value: Optional[float] = None + if confidence is not None and confidence != "": + confidence_value = float(str(confidence)) return cls( call_id=str(payload.get("call_id", "")), digits=str(payload["digits"]) if "digits" in payload else None, speech_result=str(payload["speech_result"]) if "speech_result" in payload else None, - confidence=float(confidence) if confidence not in (None, "") else None, + confidence=confidence_value, voice=str(payload["voice"]) if "voice" in payload else None, from_=str(payload["event_from"]) if "event_from" in payload else None, to=str(payload["event_to"]) if "event_to" in payload else None,