使用说明

PJSUA MCP服务器

一主控程序该服务器使AI助手能够同时控制多个SIP用户代理。建立在 PJSUA2 （pjproject 2.14.1）使用Python 3.13，打包在Docker中。

一个MCP服务器进程并行管理N部手机。每部手机都有自己的 pj.Account 并且它自己的UDP传输在一个单一的 pj.Endpoint。添加手机时，服务器会为每个手机注册22个操作工具(<phone_id>_make_call, <phone_id>_hangup，…）通过 mcp.add_tool() 和火灾 notifications/tools/list_changed当你放下手机时，那些工具又消失了。

在这些原子工具之上，服务器还附带了 事件驱动场景引擎 (src/scenario_engine/)LLM代理将一个多步骤SIP流描述为一个YAML场景，引擎在自己的异步循环中执行整个过程，而没有每一步的LLM转换延迟。场景组成原子模式（14开箱即用）或定义内联 hooks: 对于一次性流程，并返回每个事件+行动的完整时间表以供事后检查。看 场景引擎 下面的部分。

建筑

┌──────────────────────────────────────────────────────────────┐
│  AI Assistant (Claude, etc.)                                 │
│                                                              │
│  "Load the test profile, then call from a to 002"            │
└──────────────┬───────────────────────────────────────────────┘
               │ MCP (JSON-RPC over stdio)
               ▼
┌──────────────────────────────────────────────────────────────┐
│  PJSUA MCP Server (one Docker container)                     │
│                                                              │
│  ┌───────────────────── scenario engine ─────────────────┐   │
│  │ EventBus ◄── emit reg.* / call.state.* / dtmf.* / im  │   │
│  │    ▲           from pjsua callbacks                   │   │
│  │    │                                                  │   │
│  │ HookRuntime                    ActionExecutor         │   │
│  │    │                            (maps 19 actions       │   │
│  │    │                             to CallManager etc.)  │   │
│  │    │                                                   │   │
│  │ Orchestrator ──► run_scenario / validate_scenario      │   │
│  │                                                          │   │
│  └──────────────────────────────────────────────────────┘   │
│                       ▼                                      │
│  ┌────────────┐  ┌───────────────┐  ┌─────────────────────┐  │
│  │ SipEngine  │  │ PhoneRegistry │  │     CallManager     │  │
│  │ (Endpoint, │  │  dict[pid]    │  │  dict[call_id],     │  │
│  │  codecs,   │  │  → SipAccount │  │  per-phone queues,  │  │
│  │  per-phone │  │   + Config    │  │  incoming routing,  │  │
│  │  transports)│  │               │  │  always-on recording│  │
│  └──────┬─────┘  └──────┬────────┘  └──────┬──────────────┘  │
│         │               │                   │                │
│         │          ┌────┴─── phone_tool_factory ─────┐       │
│         │          │  register_phone_tools(mcp, pid) │       │
│         │          │    → 22 closures per phone      │       │
│         │          │    → mcp.add_tool / remove_tool │       │
│         │          └─────────────────────────────────┘       │
│         │                                                    │
│  ┌──────┴──────────────────────────────────────────────────┐ │
│  │              PJSUA2 / pjproject 2.14.1                  │ │
│  └──────────────────────┬──────────────────────────────────┘ │
│                         │ SIP/UDP (1 socket per phone)       │
│  ┌──────────────┐       │                                    │
│  │ SipLogWriter  │ ◄────┘  captures every SIP message        │
│  └──────────────┘                                            │
│  ┌──────────────┐                                            │
│  │ PcapManager   │  tcpdump — host-wide or BPF per phone     │
│  └──────────────┘                                            │
└──────────────────────────────────────────────────────────────┘
               │ SIP/UDP
               ▼
        ┌─────────────┐
        │  SIP PBX /   │
        │  Registrar   │
        └─────────────┘

MCP工具

静态（11-始终存在）

电话CRUD

工具	说明
`list_phones`	所有已注册的电话，包括注册状态、传输端口、活动呼叫计数、每个电话工具名称
`add_phone`	创建一个交通+SipAccount，发送REGISTER，注册22个手机操作工具
`drop_phone`	挂断电话、注销、关闭传输、卸载手机工具
`get_phone`	一部手机的完整信息——凭证（无密码）、注册状态、活动通话、， `recording_enabled`
`update_phone`	更改运行时设置-- `auto_answer` / `codecs` （每部手机SDP过滤器，即时）/ `recording_enabled` / `capture_enabled` （即时）或凭据（强制重新注册）
`load_phones`	批量添加YAML配置文件中列出的每个手机。默认情况下原子替换(`merge=True` 用于追加销售）

全球诊断

工具	说明
`get_sip_log`	检索pjsip日志条目（原始文本）。 `phone_id` 所有权过滤器（呼叫ID/传输端口/RISTER）； `call_id`/`method`/`direction`/`status_code`/`cseq` 进一步缩小； `filter_text` 子字符串转义符
`get_call_messages`	结构化SIP消息——解析的报头+解析的SDP。与相同的过滤器集 `get_sip_log` 减 `filter_text`.专为程序化计划检查而设计
`list_recordings`	步行 `/recordings/` （以及传统平面文件）用于每个WAV；筛选依据 `phone_id` / `call_id`
`analyze_capture`	解析 `/captures/<phone_id>/call_<call_id>_*.pcap` 转换为结构化RTP/RTCP流计数。表面 `phone_rtp_codecs_seen` + `non_phone_codecs_on_phone_port` （与手机相比 `codecs` config），因此调用者可以验证每个电话的SDP过滤器的一致性，而无需在bash中进行ad-hoc pcap解析

每部手机的数据包捕获仍在继续 update_phone(phone_id=..., capture_enabled=true/false) --自动在第一次音频活动呼叫时启动tcpdump，并在最后一次断开连接时停止。pcap路径在WAV中着陆 .meta.json 侧车（旁边 local_rtp_port / remote_rtp_port，在媒体处于活动状态时进行快照），因此可以进行录制、捕获和 analyze_capture 所有这些都在磁盘上配对。

场景引擎

工具	说明
`validate_scenario`	静态模拟运行——捕获未知动作、未知事件类型、格式错误的钩子——无需触摸pjsua
`run_scenario`	执行场景字典，首先自动验证，返回完整的时间线+状态+错误

每部手机动态（每部手机22部）

注册时间 add_phone （或 load_phones)将手机上网；未注册 drop_phone。以下示例使用电话 a:

工具	说明
`a_make_call`	带可选自定义SIP标头的出站INVITE
`a_answer_call`	接听电话a上的来电（如果 `call_id` 省略）
`a_reject_call`	使用SIP状态码（486/603/480）拒绝
`a_hangup`	BYE是一个正在进行的通话
`a_get_call_info`	状态、编解码器、持续时间、RTP统计数据、远程/本地联系人、录制路径
`a_get_call_history`	已完成的电话a通话
`a_list_calls`	a跟踪呼叫的简明状态摘要
`a_get_active_calls`	完整信息+RTP的主动通话
`a_send_dtmf`	在通话中发送DTMF数字
`a_hold` / `a_unhold`	重新邀请仅发送/恢复
`a_blind_transfer`	请参考以重定向a的呼叫
`a_attended_transfer`	参考+替换。双腿必须属于电话a——跨电话桥接被拒绝
`a_conference`	将多个自有电话连接到会议中
`a_play_audio` / `a_stop_audio`	将WAV播放到通话/恢复MOH中
`a_get_recording`	电话呼叫的WAV+sidecar元的路径/大小
`a_send_message` / `a_get_messages`	SIP消息发件箱/收件箱
`a_register` / `a_unregister`	新鲜寄存器循环/去寄存器（对称对）
`a_get_registration_status`	手机a的快速注册状态

N个电话的总表面：10+22·N。

快速开始

1.构建Docker镜像

docker compose build

2.连接到AI助手

添加到MCP客户端配置中（例如。 .mcp.json):

{
  "mcpServers": {
    "pjsua": {
      "command": "docker",
      "args": ["compose", "-f", "/absolute/path/to/pjsua_mcp/docker-compose.yml",
               "run", "--rm", "-i", "pjsua-mcp"]
    }
  }
}

3.描述你的手机（YAML配置文件）

服务器附带没有SIP凭据——您可以在主机上的YAML配置文件中描述您的手机。

cp config/phones.example.yaml config/phones.yaml
$EDITOR config/phones.yaml

config/phones.yaml 被忽视；仅 phones.example.yaml 被跟踪。docker组合绑定挂载 ./config → /config （只读）。

最小配置文件：

defaults:                 # optional — merged into every phone, phone-level keys win
  domain: sip.example.com
  password: change_me
  codecs: [PCMA, telephone-event]   # default for phones that don't override
  auto_answer: false

phones:
  - phone_id: a
    username: "1001"
  - phone_id: b
    username: "1002"
    codecs: [PCMU, telephone-event]   # phone-level override wins over defaults
    auto_answer: true

每部手机 codecs: SDP重写过滤器——每个传出的报价或接听此电话只会列出这些编解码器。RTP发送/接收自然会这样，因为pjsua的媒体激活从中选择编解码器 {SDP广告}∩{端点已启用}（端点固定启动时的超集）。DTMF(telephone-event)由自动保存即使没有明确列出重写器。没有手机 codecs 列表会回退到端点超集提供的任何内容。

4.加载配置文件并运行场景

mcp__pjsua__load_phones()                   # reads /config/phones.yaml
# → every phone registers; a_make_call, b_hangup, … appear via tools/list_changed.

mcp__pjsua__a_make_call(dest_uri="sip:002@sip.example.com")
mcp__pjsua__a_get_call_info(call_id=0)
mcp__pjsua__a_hangup(call_id=0)

load_phones 是 原子替代 默认情况下：在加载之前，每个现有手机的活动通话都会挂断，手机也会掉线。通过 merge=True 保留新配置文件中未列出的手机。

对于不涉及配置文件的临时添加：

mcp__pjsua__add_phone(phone_id="alice",
                      domain="sip.example.com",
                      username="1099", password="x",
                      codecs=["PCMA", "telephone-event"])
# → phone alice's INVITEs list only PCMA + telephone-event in SDP,
#   and RTP for this phone uses PCMA.
mcp__pjsua__drop_phone(phone_id="alice")

通话场景——两种风格

服务器支持两种驾驶通话模式：

原子工具模式 --直接拨打电话工具(a_make_call, b_answer_call，…）并在步骤之间轮询 time.sleep很好交互式调试和一次性实验。
场景引擎模式 --将整个流程描述为YAML场景交给 run_scenario发动机臂钩在活动总线上并在一个紧密的异步循环中驱动流，因此时序是确定性，没有每一步LLM转弯延迟。有益于可复制的测试用例和票复制器。

以下示例显示了发动机形式（首选可重复工作）；原子工具等效工具始终可用作后备。

所有示例都假设配置文件已加载(a, b, c 在线的用注册商的URI替换URI。

基本呼叫：A→ B带DTMF

# scenario
name: a-to-b-dtmf
phones: [a, b]
patterns:
  - {use: auto-answer, phone_id: b, delay_ms: 500}
  - {use: send-dtmf-on-confirmed, phone_id: a, digits: "1234"}
  - {use: hangup-after-duration, phone_id: a, duration_ms: 5000}
  - {use: make-call-and-wait-confirmed, phone_id: a,
     dest_uri: "sip:002@sip.example.com"}
stop_on: [{phone_id: a, event: call.state.disconnected}]
timeout_ms: 15000

run_scenario(scenario=<dict above>)
# Returns: {status: "ok", timeline: [...events + actions with ms offsets...]}

原子工具等效物：

a_make_call(dest_uri="sip:002@sip.example.com")
# b auto-answers (auto_answer: true in profile)
a_send_dtmf(call_id=0, digits="1234")
a_hangup(call_id=0)

自动应答（IVR/bot模式）

集 auto_answer: true 对于YAML中的手机或运行时的切换：

update_phone(phone_id="b", auto_answer=True)

…或将其构建到场景中 auto-answer 图案。

盲转：B转A→ C

name: blind-transfer
phones: [a, b, c]
patterns:
  - {use: auto-answer, phone_id: b, delay_ms: 200}
  - {use: auto-answer, phone_id: c, delay_ms: 200}
  - {use: blind-transfer, phone_id: b,
     transfer_to: "sip:003@sip.example.com", after_ms: 2000}
  - {use: hangup-after-duration, phone_id: a, duration_ms: 5000}
  - {use: make-call-and-wait-confirmed, phone_id: a,
     dest_uri: "sip:002@sip.example.com"}
stop_on: [{phone_id: c, event: call.state.disconnected}]
timeout_ms: 12000

原子工具等效物：请参见 b_blind_transfer(dest_uri=...).

参加转移：B持有A，咨询C，桥接A↔ C

流程有足够的步骤，因此更清晰 内联场景挂钩 而不是作为复合图案：

name: attended-transfer
phones: [a, b, c]
patterns:
  - {use: auto-answer, phone_id: b, delay_ms: 200}
  - {use: auto-answer, phone_id: c, delay_ms: 200}
  - {use: make-call-and-wait-confirmed, phone_id: a,
     dest_uri: "sip:002@sip.example.com"}
hooks:
  - when: call.state.confirmed
    on_phone: a
    once: true
    then:
      - wait: 1000ms
      - hold
      - make_call: {phone_id: a, to: "sip:003@sip.example.com"}
      - wait: 2500ms
      - attended_transfer
stop_on: [{phone_id: c, event: call.state.disconnected}]
timeout_ms: 15000

双腿必须属于同一部手机——跨手机参与转接返回一个错误，并显示一条明确的消息。

三方会议

name: conference
phones: [a, b, c]
patterns:
  - {use: auto-answer, phone_id: b, delay_ms: 100}
  - {use: auto-answer, phone_id: c, delay_ms: 100}
initial_actions:
  - {action: make_call, phone_id: a, to: "sip:002@sip.example.com"}
  - {action: make_call, phone_id: a, to: "sip:003@sip.example.com"}
hooks:
  - when: call.state.confirmed
    on_phone: a
    once: true
    then:
      - wait: 2500ms
      - action: conference
        phone_id: a
        call_ids: auto        # engine resolves to all active calls on a
      - wait: 5000ms
      - hangup_all: {phone_id: a}
stop_on: [{phone_id: a, event: call.state.disconnected}]
timeout_ms: 20000

编解码器选择和通话中期更改

每个手机的编解码器列表都经过SDP重写器；通话中重新邀请通过 reinvite-codec-change 模式仍然使用全局优先级：

patterns:
  - {use: reinvite-codec-change, phone_id: a, new_codec: G722,
     trigger_at_ms: 3000}
  - {use: hangup-after-duration, phone_id: a, duration_ms: 5000}
  - {use: make-call-and-wait-confirmed, phone_id: a,
     dest_uri: "sip:002@sip.example.com"}

SIP消息传递

initial_actions:
  - {action: send_message, phone_id: a,
     to: "sip:002@sip.example.com", body: "Hello!"}
stop_on: [{event: im.received, phone_id: b}]
timeout_ms: 2000

监控（始终是原子性的--只读自检）

list_phones()                # reg state + active-call counts
a_get_active_calls()         # a's active calls with RTP
a_list_calls()               # compact summary incl. DISCONNECTED
get_sip_log(phone_id="a", last_n=30)

场景引擎

目标。 让LLM编写一次多步骤SIP流并执行它确定性。发动机更换“呼叫工具，等待2秒，呼叫下一个” “工具”循环（在LLM转弯延迟上燃烧挂钟并与之竞争真正的SIP定时器）与在一个异步循环中运行的YAML流。

典型工作流程中的两个工具

validate_scenario(scenario=<dict>)     # static dry-run (no pjsua touched)
run_scenario(scenario=<dict>)          # execute and return the timeline

这两个工具都只接受一个场景作为Python字典——文件路径不是支持（代理在容器外运行，因此路径不会转换）。

场景的编写方式为 内联 hooks: — when: <event> + then: [<actions>]. 规范参考 用于解剖学、习语、全动作表面，以及工作实例（盲转、有人值守转）生活在 pjsua-scenarios 此MCP附带的技能。

事件分类

胡克在听：

呼叫状态: call.state.{calling,incoming,early,connecting,confirmed,disconnected}
双音多频: dtmf.in, dtmf.out
注册: reg.{started,success,failed,unregistered}
消息传递: im.received
场景生命周期: scenario.{started,stopped}
用户发出: user.<name> （从 emit 行动）

动作词汇（19个动作）

呼叫控制: answer, hangup, hangup_all, reject, hold, unhold, send_dtmf, blind_transfer, attended_transfer, conference, make_call
媒体: play_audio, stop_audio, send_message, set_codecs
流量控制: wait, wait_until, emit, checkpoint, log

分派时继承的默认值： phone_id 从钩子 on_phone 或触发事件， call_id 从触发事件开始。

`stop_on` 过滤器

stop_on:
  - phone_id: a
    event: call.state.disconnected
    call_id: 2                       # specific call-id
  - event: call.state.disconnected
    match: {last_status: "4xx"}      # predicate — supports exact, list,
                                     # "4xx"/"5xx", "~regex"

飞行前验证

run_scenario 自动运行 validate_scenario 第一。拼写错误（错误动作，错误的事件前缀、格式错误的钩子）返回 status="error" 在<100毫秒内-- 超时时没有挂钟烧坏。

工作示例

例子活在 pjsua-scenarios 技能（随MCP一起运送）：

.claude/skills/pjsua-scenarios/SKILL.md --解剖学、习语、简单调用模板
.claude/skills/pjsua-scenarios/examples/blind-transfer.md --参考+排序
.claude/skills/pjsua-scenarios/examples/attended-transfer.md --协调器挂钩+参考/替换

呼叫信息和RTP统计

<phone>_get_call_info 返回实时呼叫数据，包括RTCP导出的RTP统计数据：

{
  "phone_id": "a",
  "call_id": 0,
  "state": "CONFIRMED",
  "remote_uri": "sip:002@sip.example.com",
  "remote_contact": "<sip:192.0.2.10:5060;ob>",
  "local_contact": "<sip:1001@192.0.2.20:5062>",
  "codec": "PCMA",
  "duration": 45,
  "recording_file": "/recordings/a/call_0_20260101_141603_528491.wav",
  "playing_file": "/app/audio/moh.wav",
  "rtp": {
    "tx_packets": 2250, "tx_bytes": 360000,
    "rx_packets": 2248, "rx_bytes": 359680,
    "rx_loss": 0, "rx_dup": 0, "rx_reorder": 0, "rx_discard": 0,
    "rx_jitter_usec": 875, "rtt_usec": 6362
  }
}

<phone>_get_active_calls 一次为手机上的每个活动呼叫返回此值——无需迭代 call_ids

通话录音（每部手机切换，配对pcap）

录音是 默认情况下关闭 --通过手机选择加入 recording_enabled: true 在YAML或 recording_enabled=True 在 add_phone。启用后，手机上的每个通话都会写入集装箱路径 /recordings/<phone_id>/ 作为两个成对的文件：

/recordings/
├── a/
│   ├── call_0_20260422_145828_123456.wav        # local + remote audio mixed
│   └── call_0_20260422_145828_123456.meta.json  # context sidecar
├── b/
│   └── ...

文件名带有微秒后缀，因此单个调用可以产生如果在通话过程中切换录音，则需要几个WAV（见下文）。侧三轮承载着WAV本身所缺乏的背景：

{
  "phone_id": "a", "call_id": 0, "direction": "outbound",
  "started_at": "2026-04-22T14:58:28+00:00",
  "ended_at":   "2026-04-22T14:58:54+00:00",
  "duration": 26, "codec": "PCMA",
  "remote_uri": "sip:123002@...", "last_status": 200, "last_status_text": "OK",
  "recording": "/recordings/a/call_0_20260422_145828_123456.wav",
  "pcap":      "/captures/a/call_0_20260422_145828.pcap"
}

pcap 每当每个手机自动捕获时都会填充 (capture_enabled=true)在通话过程中跑步。pcap活着在...之下 /captures/<phone_id>/ 与录音同名，因此，音频和信令配对时没有任何时间戳匹配。

每部手机切换： `recording_enabled`

每部手机都有一个 recording_enabled 标志（默认 false).设置它在YAML中或运行时，切换立即生效该电话的当前通话：

# config/phones.yaml
defaults:
  domain: sip.example.com
  password: xxx
  # recording_enabled: false      # default — nobody records
phones:
  - phone_id: a
    username: "1001"
    recording_enabled: true       # per-phone opt-in
  - phone_id: b
    username: "1002"              # stays off

add_phone(phone_id="c", domain="...", username="1003", password="x",
          recording_enabled=True)                         # opt-in at add time

update_phone(phone_id="a", recording_enabled=True)        # flip on mid-call
update_phone(phone_id="a", recording_enabled=False)       # flip back off

每 off → on 使用新的微秒唯一文件名打开一个新的WAV 每一个 on → off 关闭当前WAV并写入 .meta.json 侧三轮。所以 on → off → on → off → on → hangup 生产三 WAV+侧三轮配对 /recordings/<phone_id>/，一个也没有。使用 list_recordings(phone_id=..., call_id=...) 查看每个片段给定的呼叫； <phone>_get_recording(call_id=...) 仅返回目前开放的部分。

对主机完全隐藏录制文件，放下 ./recordings 从您的 docker-compose.yml — /recordings 将存在于临时容器FS中，并随 --rm.

音乐暂停 当呼叫连接时自动播放——西班牙组曲Op.47——Leyenda（Albeniz），来自FreeSWITCH/MUSOPEN的CC0公共域，8kHz WAV。使用 <phone>_play_audio 为了覆盖， <phone>_stop_audio 恢复卫生部。

SIP日志检查

PJSUA2堆栈处理的每个SIP消息都由自定义 LogWriter 转换为有界的内存双端队列（5000个条目）：

get_sip_log()                                        # everything (all phones)
get_sip_log(last_n=20)
get_sip_log(filter_text="401")                       # raw substring escape hatch
get_sip_log(phone_id="a")                            # ownership-resolved (Call-ID + transport port + REGISTER username)
get_sip_log(phone_id="a", call_id=0)                 # narrow to one SIP dialog
get_sip_log(phone_id="a", method="INVITE")           # structured method filter
get_sip_log(phone_id="a", direction="TX")            # outgoing only
get_sip_log(phone_id="a", status_code=200)           # 200 OK responses only
get_sip_log(phone_id="a", method="INVITE", direction="TX")  # composable

电话过滤使用结构所有权而不是子字符串匹配，因此，另一部手机腿上的消息（例如bob的RX INVITE From: <sip:alice@>，或者鲍勃的 [DISCONNECTED] 显示alice URI的转储在 To:)做不泄露到爱丽丝的过滤日志中。所有者的条目无法从结构上解析，退回到子字符串匹配响应面a warning 带计数的字段。

每个条目包含：

level --pjsip日志级别（1=错误…5=跟踪）
msg --完整日志行，包括SIP消息转储
thread --发起pjlib线程名称

结构化消息(`get_call_messages`)

当您需要解析SDP/报头（编解码器列表、媒体端口、RTCP端口、，方向）而不是原始文本，使用 get_call_messages.相同的过滤器组作为 get_sip_log 减 filter_text:

get_call_messages(phone_id="a", call_id=0, method="INVITE", direction="TX")

每条SIP消息返回一个条目：

{
  "phone_id": "a",
  "messages": [
    {
      "ts": "16:05:25.556",
      "direction": "TX",
      "method": "INVITE",
      "cseq": 8398,
      "call_id": "5d0cbc47-...",
      "from": "sip:6001@asterisk",
      "to": "sip:6002@asterisk",
      "headers": {"Call-ID": "...", "CSeq": "8398 INVITE", "Content-Type": "application/sdp"},
      "sdp": {
        "version": 0,
        "origin": {"username": "-", "ip": "192.168.1.40"},
        "media": [{
          "type": "audio", "port": 4000, "protocol": "RTP/AVP",
          "payload_types": [0, 120],
          "codecs": [
            {"pt": 0, "name": "PCMU", "clock_rate": 8000},
            {"pt": 120, "name": "telephone-event", "clock_rate": 8000, "fmtp": "0-16"}
          ],
          "direction": "sendrecv", "rtcp_port": 4001
        }]
      }
    }
  ],
  "total_count": 1
}

回复还包括 status_code.SIP消息没有 application/sdp 身体有 sdp: null.pjlib库日志行和 pjsua通话转储摘要([DISCONNECTED])默默地落下-- 它们没有需要结构化的SIP信封。

数据包捕获

两种独立模式并存：手册（一枪tcpdump你开枪来自工具调用）以及 自动捕获 （每部手机 capture_enabled flag--tcpdump在第一次音频活动呼叫时打开，在最后断开连接）。两个土地下 /captures/<phone_id>/ 与相同 basename作为录音，因此pcap和WAV在磁盘上配对。

每部手机自动捕获(`capture_enabled`)

默认值为 false --除非您选择加入，否则tcpdump不会运行。请将其打开 YAML或运行时；每次通话都会检查状态，因此您可以翻转它中期会议：

# config/phones.yaml
phones:
  - phone_id: a
    username: "1001"
    capture_enabled: true        # every call on 'a' → pcap
  - phone_id: b
    username: "1002"             # inherits default → no pcap

add_phone(phone_id="c", domain="...", username="1003", password="x",
          capture_enabled=True)                         # opt-in at add time
update_phone(phone_id="a", capture_enabled=False)       # flip off mid-call
update_phone(phone_id="a", capture_enabled=True)        # flip back on

On→在实时通话中关闭会刷新并关闭当前的pcap； off→on 使用新的微秒唯一文件名打开一个新的pcap。 Off→on 做不 追溯捕获呼叫早期的数据包。

每次自动捕获都使用宽BPF滤波器 udp，所以重新邀请更改RTP端口（保持/取消保持、编解码器交换）不会丢弃任何数据包中途通话。代价是磁盘：在嘈杂的网络上，pcap增长更快如果我们只锁定一个端口。如果你需要修剪，可以拆分pcap post-hoc——见下文。

在会议中（一部电话上有两个活动电话）a 单个 pcap是留着打电话，不是每条腿一个。第一个呼叫启动它；最后的断开连接会关闭它。

事后拆分SIP和RTP

因为BPF滤波器很宽(udp)，pcap包含SIP 信令和RTP媒体交织。与分开 tshark 事后：

tshark -Y 'sip' -r captures/a/call_0_*.pcap -w sip_only.pcap
tshark -Y 'rtp' -r captures/a/call_0_*.pcap -w rtp_only.pcap

动态工具注册

load_phones / add_phone / drop_phone 呼叫 mcp.add_tool() 和 mcp.remove_tool() 在运行时。MCP服务器通过以下方式宣布更改 notifications/tools/list_changed；兼容的客户端会立即重新扫描工具列表。

# Fresh server
list_tools()                → 14 static tools
add_phone("alice", ...)     → 14 + 22 = 36 tools (alice_make_call, alice_hangup, ...)
add_phone("bob", ...)       → 14 + 22·2 = 58 tools
drop_phone("alice")         → 14 + 22 = 36 tools

这 tools_changed=True 能力是MCP协议中的选择加入；服务器通过 create_initialization_options 猴子补丁在启动。

测试

单元测试

docker compose run --rm --entrypoint pytest pjsua-mcp tests/ -m "not integration" -v

涵盖SipEngine生命周期、PhoneRegistry CRUD+双帐户隔离、CallManager查找、PcapManager、SipLogWriter以及完整 场景引擎 套件：EventBus发布/订阅+线程、HookRuntime匹配语义、通过MockCallManager的每个有线操作、TimelineRecorder偏移、覆盖每个操作/事件前缀的飞行前验证器。快速（约7秒，约125次测试），无需网络。

集成测试（自包含）

docker compose -f docker-compose.test.yml run --build --rm test-runner

在隔离的Docker网络上，每个测试类运行一个MCP服务器子进程+一个Asterisk PBX容器（ext 6001/6002/6003）。练习注册、出站/入站呼叫、盲人+有人值守的转移、会议、编解码器协商、SIP MESSAGE、拒绝、历史、YAML配置文件加载（替换与合并）、动态工具添加/删除、跨电话有人值守转移拒绝、带配对pcap的每部电话录音布局和 .meta.json 侧三轮。

全套测试大约需要2分钟（大约90次测试）。

┌──────────────────────────────────────────────────────────┐
│  Docker Compose network: sipnet                          │
│                                                          │
│  ┌──────────────────────────────────────────────────────┐│
│  │  test-runner container                               ││
│  │                                                      ││
│  │  pytest spawns ONE MCP server subprocess per test    ││
│  │  class. That server adds several phones via          ││
│  │  add_phone / load_phones and drives them:     ││
│  │                                                      ││
│  │     ┌──────────────────────────────────────┐         ││
│  │     │  MCP Server (a, b, c managed inside) │         ││
│  │     └──────────────┬───────────────────────┘         ││
│  │                    │ SIP/UDP                         ││
│  │                    ▼                                 ││
│  │          ┌──────────────────────┐                    ││
│  │          │  Asterisk PBX         │                   ││
│  │          │  ext 6001/6002/6003   │                   ││
│  │          └──────────────────────┘                    ││
│  └──────────────────────────────────────────────────────┘│
└──────────────────────────────────────────────────────────┘

发布到Harbor（或任何OCI注册处）

此仓库仅提供图像伪影。分发到客户端（包装器脚本、斜线命令、MCP配置）属于单独的 插件仓库 它按标签引用已发布的图像。

一次性设置

将注册表坐标放入 .env （见 .env.example):

HARBOR_HOST=harbor.example.corp
HARBOR_PROJECT=voip-tools
HARBOR_IMAGE=pjsua-mcp

缓存凭据一次： docker login "$HARBOR_HOST" --他们住在 ~/.docker/config.json.

发布发布（手册）

./scripts/publish.sh v0.3.0                   # builds, tags :v0.3.0 + :latest, pushes both
./scripts/publish.sh v0.3.0-rc1 --no-latest   # pre-release — keep :latest pointing at stable
./scripts/publish.sh v0.3.0 --platform linux/amd64,linux/arm64  # multi-arch via buildx

脚本是只读的，直到 docker push 跑步——手动干跑是安全的。 .dockerignore 保持构建上下文较小（不包括 captures/, recordings/, config/phones.yaml, .env，CI文件），因此没有任何秘密或笨重的东西被运送到图像层中。

客户如何消费

在插件仓库的包装脚本中：

IMAGE="${PJSUA_MCP_IMAGE:-harbor.example.corp/voip-tools/pjsua-mcp:v0.3.0}"
exec docker run -i --rm \
  --network host \
  --cap-add NET_RAW --cap-add NET_ADMIN \
  --user "$(id -u):$(id -g)" \
  -v "$CONFIG_DIR:/config:ro" \
  -v "$DATA_DIR/captures:/captures" \
  -v "$DATA_DIR/recordings:/recordings" \
  "$IMAGE"

针脚a 特定semver标签 在插件中——从不 :latest 对于生产客户来说——因此，一个破坏性的图像更改不会悄无声息地落在每个用户的机器上。

项目结构

pjsua_mcp/
├── src/
│   ├── server.py                    # MCP entry point, 19 static tool definitions, lifespan
│   ├── sip_engine.py                # Endpoint lifecycle, per-phone transport create/close, codecs
│   ├── account_manager.py           # PhoneRegistry, PhoneConfig, SipAccount (emits reg.* / im.* events)
│   ├── call_manager.py              # SipCall, per-phone queues, incoming routing (emits call.state.* / dtmf.in)
│   ├── phone_tool_factory.py        # 22 closures × N phones; add_tool / remove_tool
│   ├── sip_logger.py                # Custom LogWriter → bounded deque
│   ├── pcap_manager.py              # tcpdump subprocess management
│   └── scenario_engine/             # Event-driven scenario runtime
│       ├── event_bus.py             # Thread-safe pub/sub; wildcard subscribe; wait_for
│       ├── hook_runtime.py          # Arm hooks, match events, dispatch actions
│       ├── action_executor.py       # 19 actions → CallManager / PhoneRegistry / SipEngine
│       ├── orchestrator.py          # ScenarioRunner — arms hooks, runs initial_actions, awaits stop_on
│       ├── timeline.py              # Chronological event+action recorder with ms offsets
│       └── validator.py             # Pre-flight static checker (typos, unknown actions/events)
├── config/
│   ├── phones.example.yaml          # YAML profile template (tracked)
│   └── .gitignore                   # ignores phones.yaml (real credentials stay out of git)
├── audio/
│   └── moh.wav                      # Default MOH — CC0, FreeSWITCH/MUSOPEN
├── tests/
│   ├── conftest.py
│   ├── test_sip_engine.py
│   ├── test_sip_logger.py
│   ├── test_account_manager.py         # legacy single-account API kept compatible
│   ├── test_phone_registry.py          # multi-phone registry + two-account isolation
│   ├── test_call_manager.py
│   ├── test_pcap_manager.py
│   ├── test_integration.py             # end-to-end against Asterisk
│   ├── scenario_engine/                # ~70 unit tests for the engine
│   │   ├── test_event_bus.py
│   │   ├── test_hook_runtime.py
│   │   ├── test_orchestrator.py
│   │   ├── test_timeline.py
│   │   ├── test_validator.py
│   │   └── test_actions_direct.py
│   └── asterisk/
│       ├── Dockerfile
│       ├── pjsip.conf
│       ├── extensions.conf
│       └── modules.conf
├── scripts/
│   └── publish.sh                   # Build + tag + push image to Harbor (manual one-liner)
├── Dockerfile                       # Multi-stage: build pjproject + runtime
├── .dockerignore                    # Trim build context (ignore recordings/captures/secrets)
├── docker-compose.yml               # Mounts ./config (ro), ./recordings, ./captures
├── docker-compose.test.yml          # Asterisk + test runner on sipnet
├── .env.example                     # UID/GID + HARBOR_HOST/HARBOR_PROJECT (copy to .env)
├── requirements.txt                 # mcp[cli], PyYAML, pydantic, pytest, jinja2, jsonschema
├── pyproject.toml
└── .mcp.json                        # MCP client config for AI assistants

技术说明

Python 3.13+pjproject 2.14.1 --在多阶段Docker构建中从源代码构建。Python 3.13已删除 distutils，所以 setuptools 在构建SWIG绑定之前安装。
零音频设备 --在Docker中无头运行，没有声卡。ALSA库在运行时仍处于链接状态。
一 pj.EndpointN pj.Account --pjsua2的原生多账户模型。每部手机都有自己的UDP传输(ep.transportCreate)，因此每部手机的数据包捕获和SIP Contact端口保持不同。
来电路由 --每个 SipAccounts onIncomingCall 回叫是通过每个电话的盖子连接的 CallManager._make_incoming_handler，所以电话打到了正确的电话 _incoming_queue.
线程模型 — threadCnt=0 使用asyncio线程的手动事件循环轮询（~50次轮询/秒）。SWIG控制器回调（LogWriter）不能从执行器线程可靠地工作。
stdout保护 --C级fd 1在启动时被重定向到stderr。MCP JSON-RPC使用原始stdout fd的保存副本。防止pjlib控制台输出损坏MCP通道。
SIP日志 — consoleLevel=5 （匹配 level=5)确保全局日志级别不会被抑制。LogWriter将所有内容捕获到线程安全的有界双端队列中。
自动应答 --推迟到事件轮询循环（不在内部 onIncomingCall)以避免PJSUA2呼叫状态机问题。
录制 --每部手机 recording_enabled 标志（默认关闭--每个手机选择加入）。打开后，写信给 /recordings/<phone_id>/call_<call_id>_<ts>_<us>.wav 加上a .meta.json 带有调用上下文和配对pcap路径的sidecar（当手机正在运行捕获时）。记录器在播放器设置后连接，以避免会议桥中断，并在每次重新连接时重新连接 onCallMediaState本地+远程音频混合成一个单声道WAV。切换 recording_enabled 中途通话 update_phone 打开/关闭不同的WAV段——每个段都有自己的sidecar——因此，如果操作员想要更细粒度的捕获，一个调用可以发出多个记录。
自动捕获 --每部手机 capture_enabled 标志（默认关闭）。打开专用 tcpdump -i any udp 对第一个音频活动呼叫进行子处理，并在最后一次断开连接时将其关闭。过滤器保持宽泛，因此重新邀请RTP端口更改不会丢弃数据包；将SIP和RTP分开 tshark -Y 事后。启动/停止请求来自pj回调线程；实际的子进程启动通过基于双端队列的挂起队列在asyncio轮询循环上运行（模式与 process_auto_answers).会议（一部手机上的2+个电话）共享一个pcap，通过以下方式计算 _active_calls_by_phone.
参数 --音频播放器重新连接到新的 aud_med 端口重新INVITE（编解码器更改、会议转换）后，TX继续流动。
动态刀具注册 — tools_changed=True 通过以下方式启用功能 create_initialization_options 猴斑； ctx.session.send_tool_list_changed() 每次添加/删除手机后都会触发（每批一次 load_phones).
旧电话清理 --断开连接的呼叫将从跟踪中删除；在重新注册之前，帐户会被关闭，以防止幽灵会话。
单点故障 --一个集装箱的碰撞导致所有N部手机掉落。可用于开发/测试台。Docker编写可以 restart: unless-stopped 如果你需要韧性。
卫生部 --西班牙组曲Op.47-Leyenda（阿尔贝尼兹），古典吉他，来自FreeSWITCH/MUSOPEN的CC0公共领域。

使用说明

PJSUA MCP服务器

建筑

MCP工具

静态（11-始终存在）

电话CRUD

全球诊断

场景引擎

每部手机动态（每部手机22部）

快速开始

1.构建Docker镜像

2.连接到AI助手

3.描述你的手机（YAML配置文件）

4.加载配置文件并运行场景

通话场景——两种风格

基本呼叫：A→ B带DTMF

自动应答（IVR/bot模式）

盲转：B转A→ C

参加转移：B持有A，咨询C，桥接A↔ C

三方会议

编解码器选择和通话中期更改

SIP消息传递

监控（始终是原子性的--只读自检）

场景引擎

典型工作流程中的两个工具

事件分类

动作词汇（19个动作）

stop_on 过滤器

飞行前验证

工作示例

呼叫信息和RTP统计

通话录音（每部手机切换，配对pcap）

每部手机切换： recording_enabled

SIP日志检查

结构化消息(get_call_messages)

数据包捕获

每部手机自动捕获(capture_enabled)

事后拆分SIP和RTP

动态工具注册

测试

单元测试

集成测试（自包含）

发布到Harbor（或任何OCI注册处）

一次性设置

发布发布（手册）

客户如何消费

项目结构

技术说明

相关 MCP

iOS/macOS 构建 MCP 服务器与命令行工具

iOS模拟器交互服务器

网页截图工具

网页无障碍扫描器

ADB设备管理服务器

网页调试代理服务

`stop_on` 过滤器

每部手机切换： `recording_enabled`

结构化消息(`get_call_messages`)

每部手机自动捕获(`capture_enabled`)