feat(voice): add respondToGather to collapse the conversational-turn round trip

README.md

@@ -324,6 +324,7 @@ Kicks off a new Bandwidth Build account. Only one tool is exposed — SMS verifi
 - `updateCall` / `updateCallBxml` — redirect, hang up, or replace BXML
 - `generateBXML` — build valid BXML from a verb list
 - `respondToCallback` — queue a BXML response for an active callback (first-write-wins)
+- `respondToGather` — generate BXML from verbs and queue it as the callback response in one call; prefer this over `generateBXML` + `respondToCallback` for conversational turns, since fewer round trips means less chance of losing the race against Bandwidth's callback retry window
 - `getCallbackEvents` — read recent voice / messaging callback events
 - `configureCallbacks` — point an application's webhook URLs at this server
 

src/instructions.py

@@ -62,7 +62,7 @@
    - Conversation: `generateBXML(verbs=[{"type": "SpeakSentence", "text": "How can I help?", "voice": "julie"}])` (auto_gather=True is default, enables barge-in)
 5. **Create the call**: `createCall(accountId, from, to, applicationId, answerUrl)` where answerUrl = BW_MCP_BASE_URL + `/callbacks/voice/answer`.
 6. **Queue BXML immediately**: `respondToCallback(call_id, bxml)` — do this right after createCall. The BXML is delivered when the callee picks up.
-7. **For conversations**: Poll `getCallbackEvents(event_type="voice.gather")` for caller speech, generate new BXML, deliver with `respondToCallback`.
+7. **For conversations**: Poll `getCallbackEvents(event_type="voice.gather")` for caller speech, then respond in one call with `respondToGather(call_id, verbs)` — it generates the BXML and queues it in a single tool call, instead of a separate generateBXML + respondToCallback round trip. Every conversational turn races Bandwidth's callback retry window; each extra tool call is extra time that can lose that race and hang up the call before your response lands. Use respondToGather for every turn after the first, not just the opening line.
 
 ### BXML tips
 - `auto_gather=True` (default) wraps SpeakSentence in Gather for barge-in (caller can interrupt).
@@ -138,7 +138,7 @@
         ],
         LOOKUP_SECTION,
     ),
-    (["createCall", "generateBXML", "respondToCallback"], VOICE_SECTION),
+    (["createCall", "generateBXML", "respondToCallback", "respondToGather"], VOICE_SECTION),
     (
         ["getInboundMessages", "getCallbackEvents", "configureCallbacks"],
         CALLBACK_SECTION,

src/profiles.py

@@ -21,6 +21,7 @@
         # Custom tools
         "generateBXML",
         "respondToCallback",
+        "respondToGather",
         "getCallbackEvents",
         "configureCallbacks",
         # Discovery — find your number and app

src/specs/AGENTS.md

@@ -150,6 +150,7 @@ Auth: client_credentials. Voice application ID is required for `createCall`
 | `updateCallBxml` | Replace the BXML on an active call | poll `getCallState` |
 | `generateBXML` | Build valid BXML from a verb list | inspect returned XML before sending |
 | `respondToCallback` | Queue a BXML response for an active callback | first-write-wins; second writer gets `code: "conflict"` |
+| `respondToGather` | Generate BXML from verbs and queue it as the callback response, in one call | same first-write-wins semantics as `respondToCallback`; prefer this over `generateBXML` + `respondToCallback` for every turn after the first — fewer round trips means less chance of losing the race against Bandwidth's callback retry window |
 | `getCallbackEvents` | Read recent voice/messaging callback events | check `event_type` and `timestamp` |
 | `configureCallbacks` | Point an application's callback URLs at this server | confirm via `listApplications` |
 

src/tools/voice.py

@@ -139,6 +139,23 @@ async def respond_to_callback_flow(
     return {"status": "queued", "call_id": call_id}
 
 
+async def respond_to_gather_flow(
+    event_store: EventStore,
+    call_id: str,
+    verbs: list[dict[str, Any]],
+    auto_gather: bool = True,
+    gather_url: str = "",
+) -> dict:
+    """Generate BXML from verbs and queue it as the callback response for a
+    call, in one step. Collapses generate_bxml_flow + respond_to_callback_flow
+    into a single round trip -- see respondToGather's tool docstring for why
+    that matters (the callback-retry-window race)."""
+    bxml = await generate_bxml_flow(verbs, auto_gather, gather_url)
+    result = await respond_to_callback_flow(event_store, call_id, bxml)
+    result["bxml"] = bxml
+    return result
+
+
 def register_voice_tools(mcp, event_store: EventStore, config: dict = None) -> None:
     @mcp.tool(name="generateBXML")
     async def generate_bxml(
@@ -178,3 +195,35 @@ async def respond_to_callback(call_id: str, bxml: str) -> dict:
             bxml: Valid BXML string (use generateBXML to produce this).
         """
         return await respond_to_callback_flow(event_store, call_id, bxml)
+
+    @mcp.tool(name="respondToGather")
+    async def respond_to_gather(
+        call_id: str,
+        verbs: list[dict[str, Any]],
+        auto_gather: bool = True,
+    ) -> dict:
+        """Generate BXML from verbs and queue it as the response to an active
+        voice call, in a single tool call.
+
+        This collapses the usual two-step turn (generateBXML, then
+        respondToCallback) into one round trip. Every turn of a live
+        conversation is a race against Bandwidth's answer/gather callback
+        retry window -- each extra tool call is extra agent reasoning time
+        that can lose that race and hang up the call before the response is
+        delivered. Prefer this over generateBXML + respondToCallback whenever
+        you are responding to an active call, not just generating BXML to
+        inspect it.
+
+        Still read the caller's input from getCallbackEvents (event_type=
+        "voice.gather") first -- this tool does not read pending events for
+        you, it only merges the generate-and-queue step. First-write-wins
+        applies exactly as it does for respondToCallback.
+
+        Args:
+            call_id: The call ID to respond to.
+            verbs: List of BXML verb descriptions (same shape as generateBXML).
+            auto_gather: Wrap SpeakSentence in Gather for barge-in. Default True.
+        """
+        base_url = (config or {}).get("BW_MCP_BASE_URL", "")
+        gather_url = f"{base_url}/callbacks/voice/gather" if base_url else ""
+        return await respond_to_gather_flow(event_store, call_id, verbs, auto_gather, gather_url)

test/test_bxml.py

@@ -153,3 +153,61 @@ async def test_xml_escaping():
         [{"type": "SpeakSentence", "text": 'Use <b> & "quotes"'}]
     )
     fromstring(result)
+
+
+@pytest.mark.asyncio
+async def test_respond_to_gather_generates_and_queues_in_one_call():
+    """respondToGather is the fast path for a conversational turn: generate
+    BXML from verbs and queue it as the callback response in a single tool
+    call, instead of the separate generateBXML + respondToCallback round
+    trip. Fewer round trips means less agent reasoning time between the
+    callback landing and the response being queued -- directly addresses
+    the callback-retry-window race documented in respond_to_callback_flow's
+    real-world failure mode (an LLM agent's per-tool-call latency losing the
+    race against Bandwidth's answer/gather retry window)."""
+    from src.event_store import EventStore
+    from src.tools.voice import respond_to_gather_flow
+
+    event_store = EventStore(max_events=100, ttl_seconds=3600)
+    result = await respond_to_gather_flow(
+        event_store,
+        call_id="call-1",
+        verbs=[{"type": "SpeakSentence", "text": "Got it, one moment."}],
+        auto_gather=False,
+    )
+
+    assert result["status"] == "queued"
+    assert result["call_id"] == "call-1"
+    assert "<SpeakSentence" in result["bxml"]
+    assert "Got it, one moment." in result["bxml"]
+    fromstring(result["bxml"])
+
+    # The BXML must actually be queued on the call, not just returned --
+    # this is the whole point of collapsing the two-step flow into one call.
+    call = event_store.get_call("call-1")
+    assert call is not None
+    assert call.consume_pending_bxml() == result["bxml"]
+
+
+@pytest.mark.asyncio
+async def test_respond_to_gather_first_write_wins():
+    """Same first-write-wins semantics as respondToCallback -- a second
+    writer for the same call_id must not silently overwrite the first."""
+    from src.event_store import EventStore
+    from src.tools.voice import respond_to_gather_flow
+
+    event_store = EventStore(max_events=100, ttl_seconds=3600)
+    first = await respond_to_gather_flow(
+        event_store, call_id="call-1", verbs=[{"type": "SpeakSentence", "text": "First"}]
+    )
+    second = await respond_to_gather_flow(
+        event_store, call_id="call-1", verbs=[{"type": "SpeakSentence", "text": "Second"}]
+    )
+
+    assert first["status"] == "queued"
+    assert second.get("error") == "already_handled"
+
+    call = event_store.get_call("call-1")
+    pending = call.consume_pending_bxml()
+    assert "First" in pending
+    assert "Second" not in pending