From e65cfbe3af635dfa5d2960f3c22357fceba23c34 Mon Sep 17 00:00:00 2001 From: Rogee Date: Thu, 17 Sep 2026 15:23:46 +0800 Subject: [PATCH] support task-selectable SIP trunks --- .env.example | 2 + AGENTS.md | 8 +- agent_call/cell_worker_main.py | 10 +- agent_call/real_cell.py | 188 +++++++++++++++++++++++++-- compose.yaml | 2 + configs/ai-test.bailian.example.yaml | 2 +- configs/cell-worker.example.env | 3 + deploy/asterisk.three-sip.json | 46 +++++++ deploy/cell-routes.three-sip.json | 20 +++ deploy/mock.env.example | 2 + deploy/render_asterisk.py | 68 +++++++--- docs/部署接入_运行说明.md | 7 +- scripts/run_real_cell_call.py | 15 ++- tests/test_deployment.py | 31 +++++ tests/test_real_cell.py | 87 +++++++++++++ 15 files changed, 452 insertions(+), 39 deletions(-) create mode 100644 deploy/asterisk.three-sip.json create mode 100644 deploy/cell-routes.three-sip.json diff --git a/.env.example b/.env.example index ecdca82..4f421a4 100644 --- a/.env.example +++ b/.env.example @@ -38,6 +38,8 @@ CELL_ARI_PASSWORD= CELL_ARI_APP=agent-call CELL_SIP_TRUNK_ID=provider-primary CELL_CALLER_ID=BD93205882 +# Optional trusted JSON route map; call.execute supplies only route_policy_id/caller_profile_id. +CELL_ROUTE_MAP_JSON= CELL_RTP_BIND_HOST=127.0.0.1 CELL_RECORDING_DIR=/data/recordings diff --git a/AGENTS.md b/AGENTS.md index 129bbbe..ccdc2f2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -81,14 +81,14 @@ ## 测试呼叫号码 -- 用户指定测试被叫号码：`15003164745`（第一组）、`18625770806`（第二组）。仅用于已授权的 Mock/明确安排的测试；不得因写入此处而自动发起真实呼叫，原始号码保持不变。 +- 用户指定测试被叫号码：`15003164745`（第一组）、`18625770806`（第二组、第三组，按各自线路规则）。仅用于已授权的 Mock/明确安排的测试；不得因写入此处而自动发起真实呼叫，原始号码保持不变。 - 主叫标识保留原值（包括 `BD`），不能按纯数字手机号清洗，也不能直接当成 Digest 认证用户名；具体 From/PAI 等字段映射仍需确认。 - 业务原始被叫号码保持不变；使用该线路时按其规则构造 `7089<被叫号码>`，避免重复添加或把该前缀带到其他供应商线路。 - 传输协议、IP/Digest 鉴权、是否注册及并发限制仍需供应商确认；当前供应商已反馈需使用 PCMA，Asterisk 配置以 `allow=alaw` 表示，仍需真实线路验证。 -- 目前只提供一组服务端地址，未提供独立备用地址。不能把同一地址重复填写成主备并宣称具备容灾；现有生成器要求 primary/backup，单线路启动支持尚待调整。 +- 每条线路目前只提供一个服务端地址，未提供独立备用地址。不能把同一地址重复填写成主备并宣称具备容灾；三条已登记线路应作为独立 trunk 配置，不为凑主备虚构供应商。 -## 多 SIP 供应商方案（用户已确认，运行代码待实现） +## 多 SIP 供应商方案（用户已确认，测试运行骨架已实现） - 采用**供应商按需预接入 + 每次外呼动态选路**。先验证当前供应商，后续逐家新增，不等待所有供应商一次性接入完毕。 - 供应商/线路层预先维护独立 trunk：服务端、协议、鉴权、注册、允许的主叫、被叫改写规则和并发额度。不同供应商不能共用并逐呼覆盖同一份线路配置。 @@ -96,7 +96,7 @@ - MQ 指令不能让调用方任意注入 SIP 地址、认证凭据或越权主叫；线路选择对应的具体消息字段仍由用户制定，不擅自定接口。 - SIP Trunk、codec、主叫/被叫规则、出口池及多机器 Cell 发布由独立 Asterisk/SIP 管理后台作为唯一写入面统一管理；SaaS 仅可读取受控 Trunk 目录/状态，不直接访问 Asterisk/ARI 管理面或写入 SIP 配置。 - 切换线路时从原始被叫重新应用目标线路规则，不能沿用上一家的前缀或主叫。重试/切换条件需按业务契约另行实现，不能默认重复拨打已接通的电话。 -- 现有生成器仅支持固定 primary/backup，尚非完整多供应商路由；后续支持单线路启动、可扩展线路列表及逐呼选路，不为凑齐主备而虚构供应商。 +- Asterisk 生成器支持任意数量的独立 `trunks`；Real Cell 通过受信的 Cell 本地路由表把 `route_policy_id + caller_profile_id` 绑定到已配置 `trunk_id`、主叫和被叫前缀，任务不能注入 SIP 地址、认证凭据或主叫。无需逐呼重写 `pjsip.conf`、重载 Asterisk 或重建注册；生产路由表仍由 SIP 管理后台发布。 ## 生产容量与网络架构基线（用户已确认，运行代码待实现） diff --git a/agent_call/cell_worker_main.py b/agent_call/cell_worker_main.py index 4ceb4ec..0779c10 100644 --- a/agent_call/cell_worker_main.py +++ b/agent_call/cell_worker_main.py @@ -13,7 +13,13 @@ from typing import Any from .ai_runtime import ConversationEngine, load_prompt from .bailian import BailianASR, BailianLLM, BailianTTS, build_bailian_config from .core import PikaBroker -from .real_cell import CellCallConfig, CellExecutionLedger, RealCellCall, RealCellWorker +from .real_cell import ( + CellCallConfig, + CellExecutionLedger, + RealCellCall, + RealCellWorker, + load_cell_routes, +) def _profile(path: str) -> dict[str, Any]: @@ -77,11 +83,13 @@ def build_worker() -> RealCellWorker: ledger = CellExecutionLedger( os.environ.get("AGENT_CALL_EXECUTION_LEDGER", "/data/cell-executions.sqlite3") ) + routes = load_cell_routes(os.environ.get("CELL_ROUTE_MAP_JSON")) return RealCellWorker( broker, tenant_key, ledger, RealCellCall(CellCallConfig.from_env(), engine, asr), + routes=routes, ) diff --git a/agent_call/real_cell.py b/agent_call/real_cell.py index 72bcdf2..59c65b5 100644 --- a/agent_call/real_cell.py +++ b/agent_call/real_cell.py @@ -48,6 +48,87 @@ class CellCallError(RuntimeError): self.retryable = retryable +@dataclass(frozen=True) +class CellRoute: + """A trusted local route binding; call payloads cannot define its values.""" + + route_policy_id: str + caller_profile_id: str + trunk_id: str + caller_id: str + dial_prefix: str = "" + + def __post_init__(self) -> None: + for field_name, value in ( + ("route_policy_id", self.route_policy_id), + ("caller_profile_id", self.caller_profile_id), + ): + if not isinstance(value, str) or not re.fullmatch( + r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}", value + ): + raise CellCallError( + "ROUTE_CONFIG_INVALID", f"{field_name} is invalid" + ) + if not isinstance(self.trunk_id, str) or not re.fullmatch( + r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", self.trunk_id + ): + raise CellCallError("ROUTE_CONFIG_INVALID", "trunk_id is invalid") + if not isinstance(self.caller_id, str) or not re.fullmatch( + r"[A-Za-z0-9_.+\- ]+(?:<[A-Za-z0-9_.+\- ]+>)?", self.caller_id + ): + raise CellCallError("ROUTE_CONFIG_INVALID", "caller_id is invalid") + if not isinstance(self.dial_prefix, str) or len(self.dial_prefix) > 32 or ( + self.dial_prefix and not re.fullmatch(r"[A-Za-z0-9]+", self.dial_prefix) + ): + raise CellCallError("ROUTE_CONFIG_INVALID", "dial_prefix is invalid") + + +def load_cell_routes(raw: str | None) -> dict[str, CellRoute]: + if not raw or not raw.strip(): + return {} + try: + data = json.loads(raw) + except (TypeError, json.JSONDecodeError) as exc: + raise CellCallError("ROUTE_CONFIG_INVALID", "CELL_ROUTE_MAP_JSON is invalid") from exc + if not isinstance(data, Mapping): + raise CellCallError("ROUTE_CONFIG_INVALID", "CELL_ROUTE_MAP_JSON must be an object") + routes: dict[str, CellRoute] = {} + for route_policy_id, value in data.items(): + if not isinstance(value, Mapping): + raise CellCallError( + "ROUTE_CONFIG_INVALID", f"route {route_policy_id!r} must be an object" + ) + caller_profile_id = value.get("caller_profile_id") + trunk_id = value.get("trunk_id") + caller_id = value.get("caller_id") + dial_prefix = value.get("dial_prefix", "") + if not isinstance(caller_profile_id, str): + raise CellCallError( + "ROUTE_CONFIG_INVALID", f"route {route_policy_id!r} has invalid caller profile" + ) + if not isinstance(trunk_id, str): + raise CellCallError( + "ROUTE_CONFIG_INVALID", f"route {route_policy_id!r} has invalid trunk" + ) + if not isinstance(caller_id, str): + raise CellCallError( + "ROUTE_CONFIG_INVALID", f"route {route_policy_id!r} has invalid caller" + ) + if not isinstance(dial_prefix, str): + raise CellCallError( + "ROUTE_CONFIG_INVALID", f"route {route_policy_id!r} has invalid prefix" + ) + route = CellRoute( + route_policy_id=str(route_policy_id), + caller_profile_id=caller_profile_id, + trunk_id=trunk_id, + caller_id=caller_id, + dial_prefix=dial_prefix, + ) + routes[route.route_policy_id] = route + return routes + + @dataclass(frozen=True) class CellCallConfig: """Trusted, Cell-local settings; none are read from ``call.execute``.""" @@ -406,6 +487,10 @@ class CellCallResult: media_packets_out: int = 0 started_at: float = 0.0 ended_at: float = 0.0 + route_policy_id: str | None = None + caller_profile_id: str | None = None + trunk_id: str | None = None + effective_target: str | None = None def as_dict(self) -> dict[str, Any]: return { @@ -418,11 +503,15 @@ class CellCallResult: "media_packets_in": self.media_packets_in, "media_packets_out": self.media_packets_out, "duration_ms": max(0.0, (self.ended_at - self.started_at) * 1000), + "route_policy_id": self.route_policy_id, + "caller_profile_id": self.caller_profile_id, + "trunk_id": self.trunk_id, + "effective_target": self.effective_target, } class RealCellCall: - """Execute one authorized call using a fixed Cell-local trunk.""" + """Execute one authorized call using a trusted Cell-local route binding.""" def __init__( self, config: CellCallConfig, engine: ConversationEngine, asr: Any @@ -463,11 +552,16 @@ class RealCellCall: self._in_packets = 0 self._out_bytes = 0 - def start_authorized_call(self, callee: str) -> CellCallResult: + def start_authorized_call( + self, callee: str, route: CellRoute | None = None + ) -> CellCallResult: if not callee.isdigit() or not 3 <= len(callee) <= 32: raise CellCallError( "CALLEE_INVALID", "callee must be the original digit string" ) + selected_route = route or CellRoute( + "legacy", "legacy", self.config.trunk_id, self.config.caller_id, self.config.dial_prefix + ) self._reset_call() self.engine.history.clear() result = CellCallResult( @@ -475,6 +569,10 @@ class RealCellCall: "failed", "CELL_EXECUTOR_EXCEPTION", started_at=time.monotonic(), + route_policy_id=selected_route.route_policy_id, + caller_profile_id=selected_route.caller_profile_id, + trunk_id=selected_route.trunk_id, + effective_target=selected_route.dial_prefix + callee, ) try: self.media = RTPMedia(self.config.rtp_bind_host, self.config.rtp_bind_port) @@ -499,10 +597,10 @@ class RealCellCall: "POST", "channels", { - "endpoint": f"PJSIP/{self.config.dial_prefix + callee}@{self.config.trunk_id}", + "endpoint": f"PJSIP/{selected_route.dial_prefix + callee}@{selected_route.trunk_id}", "app": self.config.ari_app, "appArgs": self.call_id, - "callerId": self.config.caller_id, + "callerId": selected_route.caller_id, "timeout": 30, "channelId": f"target_{self.call_id}", }, @@ -1025,6 +1123,7 @@ class RealCellWorker: tenant_key: str, ledger: CellExecutionLedger, executor: RealCellCall, + routes: Mapping[str, CellRoute] | None = None, ) -> None: if not tenant_key: raise CellCallError("TENANT_KEY_INVALID", "tenant_key is required") @@ -1033,6 +1132,7 @@ class RealCellWorker: self.queue_name = f"agent-call.executor.{tenant_key}.v1" self.ledger = ledger self.executor = executor + self.routes = dict(routes or {}) self.broker.declare_tenant(tenant_key) def process_once(self) -> dict[str, Any] | None: @@ -1041,10 +1141,22 @@ class RealCellWorker: return None body = message.get("body") if isinstance(message, Mapping) else None try: - execution_id, tenant_id, agent_version_id, callee = self._command(body) + ( + execution_id, + tenant_id, + agent_version_id, + callee, + route_policy_id, + caller_profile_id, + ) = self._command(body) except CellCallError: self.broker.reject(message) return {"status": "rejected", "reason_code": "COMMAND_INVALID"} + try: + route = self._route(route_policy_id, caller_profile_id) + except CellCallError as exc: + self.broker.reject(message) + return {"status": "rejected", "reason_code": exc.code} claim = self.ledger.claim(execution_id, callee) if claim["in_doubt"]: self.broker.ack(message) @@ -1057,7 +1169,11 @@ class RealCellWorker: "AGENT_VERSION_UNAVAILABLE", "agent version is not installed on this Cell", ) - result = self.executor.start_authorized_call(callee).as_dict() + result = ( + self.executor.start_authorized_call(callee, route) + if route is not None + else self.executor.start_authorized_call(callee) + ).as_dict() except ( CellCallError, AIProviderError, @@ -1089,7 +1205,42 @@ class RealCellWorker: self.broker.ack(message) return event - def _command(self, body: Any) -> tuple[str, str, str, str]: + def _route( + self, route_policy_id: str, caller_profile_id: str + ) -> CellRoute | None: + if not route_policy_id and not caller_profile_id: + if self.routes: + raise CellCallError( + "ROUTE_POLICY_REQUIRED", "route policy is required on a multi-trunk Cell" + ) + return None + if not route_policy_id or not caller_profile_id: + raise CellCallError( + "ROUTE_POLICY_INVALID", "route policy and caller profile must be paired" + ) + if not self.routes: + if (route_policy_id, caller_profile_id) == ( + "route_policy_test", + "caller_profile_test", + ): + return None + raise CellCallError( + "ROUTE_POLICY_UNAVAILABLE", "route policy is not installed on this Cell" + ) + route = self.routes.get(route_policy_id) + if route is None: + raise CellCallError( + "ROUTE_POLICY_UNAVAILABLE", "route policy is not installed on this Cell" + ) + if route.caller_profile_id != caller_profile_id: + raise CellCallError( + "CALLER_PROFILE_MISMATCH", "caller profile is not bound to the route policy" + ) + return route + + def _command( + self, body: Any + ) -> tuple[str, str, str, str, str, str]: if not isinstance(body, Mapping) or body.get("command_type") != "call.execute": raise CellCallError("COMMAND_INVALID", "message is not call.execute") if body.get("tenant_key") != self.tenant_key: @@ -1105,11 +1256,30 @@ class RealCellWorker: payload.get("agent_version_id"), payload.get("callee"), ) + route_policy_id = payload.get("route_policy_id", "") + caller_profile_id = payload.get("caller_profile_id", "") + if not isinstance(route_policy_id, str) or not isinstance( + caller_profile_id, str + ): + raise CellCallError( + "COMMAND_INVALID", "route policy and caller profile must be strings" + ) + if bool(route_policy_id) != bool(caller_profile_id): + raise CellCallError( + "COMMAND_INVALID", "route policy and caller profile must be paired" + ) if not all(isinstance(value, str) and value for value in values): raise CellCallError( "COMMAND_INVALID", "call.execute identity fields are required" ) - return str(values[1]), str(values[0]), str(values[2]), str(values[3]) + return ( + str(values[1]), + str(values[0]), + str(values[2]), + str(values[3]), + route_policy_id, + caller_profile_id, + ) @staticmethod def _event( @@ -1134,9 +1304,11 @@ __all__ = [ "CellCallError", "CellCallResult", "CellExecutionLedger", + "CellRoute", "RTPMedia", "RealCellCall", "RealCellWorker", "alaw_to_pcm16", + "load_cell_routes", "voice_level", ] diff --git a/compose.yaml b/compose.yaml index c70f0f2..c14d84a 100644 --- a/compose.yaml +++ b/compose.yaml @@ -95,6 +95,8 @@ services: CELL_ARI_APP: "${CELL_ARI_APP:-agent-call}" CELL_SIP_TRUNK_ID: "${CELL_SIP_TRUNK_ID:?Set CELL_SIP_TRUNK_ID}" CELL_CALLER_ID: "${CELL_CALLER_ID:?Set CELL_CALLER_ID}" + CELL_DIAL_PREFIX: "${CELL_DIAL_PREFIX:-7089}" + CELL_ROUTE_MAP_JSON: "${CELL_ROUTE_MAP_JSON:-}" CELL_RTP_BIND_HOST: "${CELL_RTP_BIND_HOST:-127.0.0.1}" CELL_RECORDING_DIR: "${CELL_RECORDING_DIR:-/data/recordings}" volumes: diff --git a/configs/ai-test.bailian.example.yaml b/configs/ai-test.bailian.example.yaml index 003cb2b..a568c66 100644 --- a/configs/ai-test.bailian.example.yaml +++ b/configs/ai-test.bailian.example.yaml @@ -17,7 +17,7 @@ "max_turns": 3, "max_duration_ms": 180000, "callee": "15003164745", - "authorized_callees": ["15003164745"], + "authorized_callees": ["15003164745", "18625770806"], "max_prompt_bytes": 32768, "notes": "Use BAILIAN_* environment variables; keep secrets out of this file." } diff --git a/configs/cell-worker.example.env b/configs/cell-worker.example.env index 79fc01f..34c4e2a 100644 --- a/configs/cell-worker.example.env +++ b/configs/cell-worker.example.env @@ -23,5 +23,8 @@ CELL_ARI_PASSWORD= CELL_ARI_APP=agent-call CELL_SIP_TRUNK_ID=provider-primary CELL_CALLER_ID=BD93205882 +CELL_DIAL_PREFIX=7089 +# Optional trusted JSON route map; do not populate from an untrusted call payload. +CELL_ROUTE_MAP_JSON= CELL_RTP_BIND_HOST=127.0.0.1 CELL_RECORDING_DIR=/data/recordings diff --git a/deploy/asterisk.three-sip.json b/deploy/asterisk.three-sip.json new file mode 100644 index 0000000..78c7a23 --- /dev/null +++ b/deploy/asterisk.three-sip.json @@ -0,0 +1,46 @@ +{ + "public_ip": "123.56.71.98", + "transport": "udp", + "local_net": "172.16.0.0/12", + "trunks": [ + { + "trunk_id": "provider-primary", + "host": "61.132.228.221", + "port": 5060, + "auth_mode": "ip", + "register": false, + "from_user": "BD93205882", + "caller_id": "BD93205882", + "codec_profile": { + "allowed": ["PCMA"], + "preferred": "PCMA" + } + }, + { + "trunk_id": "provider-second", + "host": "60.171.24.90", + "port": 5060, + "auth_mode": "ip", + "register": false, + "from_user": "mbkq", + "caller_id": "mbkq", + "codec_profile": { + "allowed": ["PCMA"], + "preferred": "PCMA" + } + }, + { + "trunk_id": "provider-third", + "host": "160.202.254.79", + "port": 5060, + "auth_mode": "ip", + "register": false, + "from_user": "KQ91526", + "caller_id": "KQ91526", + "codec_profile": { + "allowed": ["PCMA"], + "preferred": "PCMA" + } + } + ] +} diff --git a/deploy/cell-routes.three-sip.json b/deploy/cell-routes.three-sip.json new file mode 100644 index 0000000..9a234b8 --- /dev/null +++ b/deploy/cell-routes.three-sip.json @@ -0,0 +1,20 @@ +{ + "route_policy_sip_first": { + "caller_profile_id": "caller_profile_sip_first", + "trunk_id": "provider-primary", + "caller_id": "BD93205882", + "dial_prefix": "7089" + }, + "route_policy_sip_second": { + "caller_profile_id": "caller_profile_sip_second", + "trunk_id": "provider-second", + "caller_id": "mbkq", + "dial_prefix": "" + }, + "route_policy_sip_third": { + "caller_profile_id": "caller_profile_sip_third", + "trunk_id": "provider-third", + "caller_id": "KQ91526", + "dial_prefix": "mka755" + } +} diff --git a/deploy/mock.env.example b/deploy/mock.env.example index e2e14ba..fad5ea7 100644 --- a/deploy/mock.env.example +++ b/deploy/mock.env.example @@ -32,5 +32,7 @@ CELL_ARI_PASSWORD=replace-before-start CELL_ARI_APP=agent-call CELL_SIP_TRUNK_ID=provider-primary CELL_CALLER_ID=BD93205882 +# Optional trusted JSON route map for a real Cell; leave empty in mock mode. +CELL_ROUTE_MAP_JSON= CELL_RTP_BIND_HOST=127.0.0.1 CELL_RECORDING_DIR=/data/recordings diff --git a/deploy/render_asterisk.py b/deploy/render_asterisk.py index 206cd79..801b77f 100755 --- a/deploy/render_asterisk.py +++ b/deploy/render_asterisk.py @@ -13,6 +13,7 @@ from pathlib import Path PUBLIC_IP = "123.56.71.98" CODEC_TO_ASTERISK = {"PCMA": "alaw", "PCMU": "ulaw"} +TRUNK_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}") def scalar(value, name, secret=False): @@ -115,6 +116,47 @@ def endpoint(name, data, env): return text + auth +def _trunks(cfg): + if "trunks" not in cfg: + primary_data = cfg.get("primary", {}) + if not isinstance(primary_data, dict): + raise ValueError("primary must be an object") + rows = [("provider-primary", primary_data)] + backup_data = cfg.get("backup") + if backup_data is not None and not isinstance(backup_data, dict): + raise ValueError("backup must be an object when configured") + if backup_data: + rows.append(("provider-backup", backup_data)) + if (primary_data.get("host"), primary_data.get("port", 5060)) == ( + backup_data.get("host"), + backup_data.get("port", 5060), + ): + raise ValueError( + "primary and backup targets must be distinct; shared failure domains still require validation" + ) + return rows + if "primary" in cfg or "backup" in cfg: + raise ValueError("use trunks instead of mixing arbitrary trunks with primary/backup") + raw = cfg["trunks"] + if not isinstance(raw, list) or not raw: + raise ValueError("trunks must be a non-empty list") + rows = [] + seen = set() + for index, data in enumerate(raw): + if not isinstance(data, dict): + raise TypeError(f"trunks[{index}] must be an object") + name = data.get("trunk_id") + if not isinstance(name, str) or not TRUNK_ID_RE.fullmatch(name): + raise ValueError( + f"trunks[{index}].trunk_id must be a safe Asterisk endpoint identifier" + ) + if name in seen: + raise ValueError(f"duplicate trunk_id: {name}") + seen.add(name) + rows.append((name, data)) + return rows + + def render(cfg, env): if cfg.get("public_ip") != PUBLIC_IP or cfg.get("transport") != "udp": raise ValueError( @@ -138,32 +180,20 @@ def render(cfg, env): password = scalar(env.get("ARI_PASSWORD"), "ARI_PASSWORD", secret=True) if len(password) < 32: raise ValueError("ARI_PASSWORD must have at least 32 characters") - primary_data = cfg.get("primary", {}) - primary = endpoint("provider-primary", primary_data, env) - backup_data = cfg.get("backup") - if backup_data is not None and not isinstance(backup_data, dict): - raise ValueError("backup must be an object when configured") - sections = [primary] - if backup_data: - backup = endpoint("provider-backup", backup_data, env) - if (primary_data["host"], primary_data.get("port", 5060)) == ( - backup_data["host"], - backup_data.get("port", 5060), - ): - raise ValueError( - "primary and backup targets must be distinct; shared failure domains still require validation" - ) - sections.append(backup) + trunks = _trunks(cfg) + sections = [endpoint(name, data, env) for name, data in trunks] + primary_name, primary_data = trunks[0] transport = f"[global]\ntype=global\nuser_agent=agent-call\n\n[transport-udp]\ntype=transport\nprotocol=udp\nbind=0.0.0.0:5060\nlocal_net={network}\nexternal_signaling_address={PUBLIC_IP}\nexternal_media_address={PUBLIC_IP}\n\n" caller = primary_data.get("caller_id") or primary_data.get("from_user") extensions = "[deny-inbound]\nexten => s,1,Hangup()\nexten => _.,1,Hangup()\n" if caller: - caller = scalar(caller, "primary.caller_id") + caller = scalar(caller, f"{primary_name}.caller_id") extensions += ( "\n[outbound]\n" - "exten => _X.,1,NoOp(agent-call outbound)\n" + "; ARI Cell calls select an approved endpoint with PJSIP/@.\n" + "exten => _X.,1,NoOp(agent-call outbound fallback)\n" f" same => n,Set(CALLERID(all)={caller})\n" - " same => n,Dial(PJSIP/${EXTEN}@provider-primary,60)\n" + f" same => n,Dial(PJSIP/${{EXTEN}}@{primary_name},60)\n" " same => n,Hangup()\n" ) return { diff --git a/docs/部署接入_运行说明.md b/docs/部署接入_运行说明.md index 4f28c46..7f381b7 100644 --- a/docs/部署接入_运行说明.md +++ b/docs/部署接入_运行说明.md @@ -202,12 +202,13 @@ if [ ! -e .local/asterisk.json ]; then fi ``` -填写实际`local_net`、主/备SIP服务器、端口、IP或Digest鉴权方式、用户名及是否注册。当前生成器按供应商反馈提供经过明确限制的UDP/PCMA外呼基线（Asterisk `allow=alaw`），TCP/TLS及特殊号码/主叫策略需要另行确认。 +填写实际`local_net`及独立SIP线路列表。三线路测试配置见`deploy/asterisk.three-sip.json`，每个`trunk_id`对应一个独立PJSIP endpoint；不能把不同供应商合并成主/备后逐呼覆盖。当前生成器按供应商反馈提供经过明确限制的UDP/PCMA外呼基线（Asterisk `allow=alaw`），TCP/TLS及特殊号码/主叫策略需要另行确认。 通过受控进程环境注入： - `ARI_PASSWORD`：至少32位，不能使用样例密码。 -- Digest接入另需`SIP_PRIMARY_PASSWORD`、`SIP_BACKUP_PASSWORD`；仅在相应接入启用Digest时要求。 +- Digest接入另需对应`SIP__PASSWORD`环境变量；仅在相应接入启用Digest时要求。 +- Real Cell 多线路运行时，将批准的`deploy/cell-routes.three-sip.json`序列化到`CELL_ROUTE_MAP_JSON`；MQ任务只携带`route_policy_id`和`caller_profile_id`，Cell拒绝未安装或主叫绑定不匹配的路由。 - 为防INI注入，换行、分号、方括号等字符会被拒绝；供应商固定密码不满足时需评审正确转义方案，不能删除校验绕过。 ```bash @@ -241,7 +242,7 @@ docker compose -f compose.asterisk.yaml exec -T asterisk \ asterisk -rx 'pjsip show contacts' ``` -只有一个节点采用host network；不要在同一宿主机直接启动第二套争用5060/8088/RTP的配置。主备trunk只是接入配置，自动FALLBACK、长期ARI事件连接、业务状态机和录音上传程序仍需开发。 +只有一个节点采用host network；不要在同一宿主机直接启动第二套争用5060/8088/RTP的配置。三条trunk是独立接入配置，任务通过受信路由表选定具体endpoint；自动FALLBACK、长期ARI事件连接、业务状态机和录音上传程序仍需按契约实现。 开始真实外呼前，必须通过SIP供应商侧日志核对我方实际出口确为123.56.71.98；仅EIP绑定成功不能证明没有其他NAT/路由改变出口。 diff --git a/scripts/run_real_cell_call.py b/scripts/run_real_cell_call.py index 5398285..c2f3f4b 100755 --- a/scripts/run_real_cell_call.py +++ b/scripts/run_real_cell_call.py @@ -50,6 +50,8 @@ def _args() -> argparse.Namespace: parser.add_argument("--callee", required=True) parser.add_argument("--tenant-id", default=None) parser.add_argument("--tenant-key", default=None) + parser.add_argument("--route-policy-id", default="route_policy_test") + parser.add_argument("--caller-profile-id", default="caller_profile_test") parser.add_argument("--broker-url", default=None) parser.add_argument("--ledger", default="/data/agent-call-cell.sqlite3") parser.add_argument("--wait-seconds", type=int, default=240) @@ -84,8 +86,8 @@ def _command( "task_item_id": f"item_{execution_id}", "task_revision": 1, "callee": args.callee, - "route_policy_id": "route_policy_test", - "caller_profile_id": "caller_profile_test", + "route_policy_id": args.route_policy_id, + "caller_profile_id": args.caller_profile_id, "agent_version_id": agent_version_id, "variables": {}, "ring_timeout_ms": 30000, @@ -142,6 +144,7 @@ def run(args: argparse.Namespace) -> dict[str, Any]: CellExecutionLedger, RealCellCall, RealCellWorker, + load_cell_routes, ) settings = _data(_path(args.profile)) @@ -181,7 +184,13 @@ def run(args: argparse.Namespace) -> dict[str, Any]: broker = PikaBroker(args.broker_url or os.environ.get("RABBITMQ_URL", "")) ledger = CellExecutionLedger(args.ledger) executor = RealCellCall(CellCallConfig.from_env(), engine, BailianASR.from_env()) - worker = RealCellWorker(broker, tenant_key, ledger, executor) + worker = RealCellWorker( + broker, + tenant_key, + ledger, + executor, + routes=load_cell_routes(os.environ.get("CELL_ROUTE_MAP_JSON")), + ) command = _command(settings, args, tenant_id, tenant_key, agent_version_id) route = f"agent-call.tenant.{tenant_key}.call.execute" broker.publish( diff --git a/tests/test_deployment.py b/tests/test_deployment.py index f65b579..fde9d38 100644 --- a/tests/test_deployment.py +++ b/tests/test_deployment.py @@ -308,6 +308,37 @@ class AsteriskTests(unittest.TestCase): self.assertIn("context=deny-inbound", files["pjsip.conf"]) self.assertIn("strictrtp=yes", files["rtp.conf"]) + def test_renders_arbitrary_task_selectable_trunks(self): + cfg = self.cfg() + cfg.pop("primary") + cfg.pop("backup") + cfg["trunks"] = [ + { + "trunk_id": "provider-first", + "host": "61.132.228.221", + "auth_mode": "ip", + "caller_id": "BD93205882", + "codec_profile": {"allowed": ["PCMA"], "preferred": "PCMA"}, + }, + { + "trunk_id": "provider-second", + "host": "60.171.24.90", + "auth_mode": "ip", + "codec_profile": {"allowed": ["PCMA"], "preferred": "PCMA"}, + }, + { + "trunk_id": "provider-third", + "host": "160.202.254.79", + "auth_mode": "ip", + "codec_profile": {"allowed": ["PCMA"], "preferred": "PCMA"}, + }, + ] + files = ast.render(cfg, {"ARI_PASSWORD": "x" * 32}) + for trunk_id in ("provider-first", "provider-second", "provider-third"): + self.assertIn(f"[{trunk_id}]", files["pjsip.conf"]) + self.assertNotIn("provider-backup", files["pjsip.conf"]) + self.assertIn("PJSIP/${EXTEN}@provider-first", files["extensions.conf"]) + def test_preserves_explicit_caller_mapping(self): cfg = self.cfg() cfg["primary"].update( diff --git a/tests/test_real_cell.py b/tests/test_real_cell.py index 862a7d4..1d636ee 100644 --- a/tests/test_real_cell.py +++ b/tests/test_real_cell.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import socket import struct import tempfile @@ -12,10 +13,12 @@ from agent_call.real_cell import ( CellCallConfig, CellCallError, CellExecutionLedger, + CellRoute, RealCellCall, RealCellWorker, RTPMedia, alaw_to_pcm16, + load_cell_routes, voice_level, ) @@ -243,6 +246,90 @@ class RealCellTests(unittest.TestCase): "mka755", ) + def test_route_map_binds_task_policy_to_trusted_trunk(self) -> None: + routes = load_cell_routes( + json.dumps( + { + "route-a": { + "caller_profile_id": "caller-a", + "trunk_id": "provider-second", + "caller_id": "mbkq", + "dial_prefix": "", + } + } + ) + ) + self.assertEqual(routes["route-a"].trunk_id, "provider-second") + self.assertEqual(routes["route-a"].caller_id, "mbkq") + with self.assertRaises(CellCallError): + load_cell_routes( + json.dumps( + { + "route-a": { + "caller_profile_id": "caller-a", + "trunk_id": "provider-second", + "caller_id": "mbkq\nspoof", + "dial_prefix": "", + } + } + ) + ) + + def test_worker_selects_installed_route_without_payload_sip_values(self) -> None: + command = { + "body": { + "command_type": "call.execute", + "tenant_id": "tenant-demo", + "tenant_key": "tenant-key", + "payload": { + "execution_id": "exec-route-1", + "agent_version_id": "agent_v1", + "callee": "18625770806", + "route_policy_id": "route-a", + "caller_profile_id": "caller-a", + }, + } + } + broker = FakeBroker([command]) + with tempfile.TemporaryDirectory() as directory: + ledger = CellExecutionLedger(Path(directory) / "ledger.sqlite3") + fake_result = SimpleNamespace( + as_dict=lambda: {"call_id": "call-route", "status": "failed", "turns": []} + ) + + class FakeExecutor: + def __init__(self) -> None: + self.engine = SimpleNamespace( + config={"agent_version_id": "agent_v1"} + ) + self.calls: list[tuple[str, CellRoute | None]] = [] + + def start_authorized_call( + self, callee: str, route: CellRoute | None = None + ) -> SimpleNamespace: + self.calls.append((callee, route)) + return fake_result + + fake_executor = FakeExecutor() + worker = RealCellWorker( + broker, + "tenant-key", + ledger, + cast(RealCellCall, fake_executor), + routes={ + "route-a": CellRoute( + "route-a", "caller-a", "provider-second", "mbkq", "" + ) + }, + ) + event = worker.process_once() + if event is None: + self.fail("worker did not publish a call.finished event") + self.assertEqual(fake_executor.calls[0][0], "18625770806") + selected_route = fake_executor.calls[0][1] + assert selected_route is not None + self.assertEqual(selected_route.trunk_id, "provider-second") + def test_ledger_marks_in_progress_as_in_doubt(self) -> None: with tempfile.TemporaryDirectory() as directory: ledger = CellExecutionLedger(Path(directory) / "ledger.sqlite3")