Natera 如何用 Amazon Bedrock AgentCore 构建低延迟语音预约智能体

python -m venv .venv
source .venv/bin/activate
pip install websockets
export AGENT_WS_URL='wss://your-agent-runtime.example/ws'
python bridge.py

# bridge.py
import asyncio
import json
import os
import uuid

import websockets

AGENT_WS_URL = os.environ["AGENT_WS_URL"]


async def relay_voice_to_agent(voice_ws, agent_ws, session_id):
    async for raw in voice_ws:
        event = json.loads(raw)

        if event.get("type") == "audio":
            # 立即确认已收到音频，不必等待完整的模型响应。
            await voice_ws.send(json.dumps({
                "type": "activity",
                "state": "processing",
                "session_id": session_id,
            }))

        event["session_id"] = session_id
        await agent_ws.send(json.dumps(event))


async def relay_agent_to_voice(agent_ws, voice_ws):
    async for raw in agent_ws:
        event = json.loads(raw)

        # 工具参数不应直接播放给患者，也不应写入普通音频日志。
        if event.get("type") == "tool_call":
            continue

        await voice_ws.send(json.dumps(event))


async def handle_call(voice_ws):
    session_id = str(uuid.uuid4())
    async with websockets.connect(
        AGENT_WS_URL,
        additional_headers={"x-session-id": session_id},
        max_queue=32,
    ) as agent_ws:
        tasks = [
            asyncio.create_task(
                relay_voice_to_agent(voice_ws, agent_ws, session_id)
            ),
            asyncio.create_task(relay_agent_to_voice(agent_ws, voice_ws)),
        ]
        done, pending = await asyncio.wait(
            tasks, return_when=asyncio.FIRST_COMPLETED
        )
        for task in pending:
            task.cancel()
        await asyncio.gather(*pending, return_exceptions=True)
        for task in done:
            task.result()


async def main():
    async with websockets.serve(handle_call, "0.0.0.0", 8765):
        print("Voice bridge listening on ws://localhost:8765")
        await asyncio.Future()


if __name__ == "__main__":
    asyncio.run(main())

from dataclasses import dataclass
from enum import IntEnum


class Trust(IntEnum):
    ANONYMOUS = 0
    IDENTIFIED = 1
    VERIFIED = 2


REQUIRED_TRUST = {
    "search_slots": Trust.IDENTIFIED,
    "book_appointment": Trust.VERIFIED,
    "cancel_appointment": Trust.VERIFIED,
}


@dataclass
class Session:
    session_id: str
    trust: Trust = Trust.ANONYMOUS


def authorize_tool(session: Session, tool_name: str) -> None:
    required = REQUIRED_TRUST.get(tool_name)
    if required is None:
        raise ValueError(f"Unknown tool: {tool_name}")
    if session.trust < required:
        raise PermissionError(
            f"{tool_name} requires {required.name}; "
            f"session has {session.trust.name}"
        )


def invoke_tool(session: Session, tool_name: str, arguments: dict):
    authorize_tool(session, tool_name)
    return {
        "ok": True,
        "tool": tool_name,
        "arguments": arguments,
        "idempotency_key": f"{session.session_id}:{tool_name}:1",
    }


if __name__ == "__main__":
    call = Session("call-42", Trust.IDENTIFIED)

    print(invoke_tool(call, "search_slots", {"postal_code": "98101"}))

    try:
        invoke_tool(call, "book_appointment", {"slot_id": "slot-9"})
    except PermissionError as exc:
        print({"ok": False, "error": str(exc)})

    call.trust = Trust.VERIFIED
    print(invoke_tool(call, "book_appointment", {"slot_id": "slot-9"}))