diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index f5bb05d..0d211f4 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -52,6 +52,7 @@ jobs:
           - hd-audio
           - cli
           - pygments
+          - mcp
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v6
diff --git a/docs/mcp.md b/docs/mcp.md
new file mode 100644
index 0000000..fa70e7e
--- /dev/null
+++ b/docs/mcp.md
@@ -0,0 +1,38 @@
+# MCP Server
+
+The `voip` package ships a ready-made [Model Context Protocol (MCP)][mcp] server
+that exposes tools to make phone calls on your behalf to any MCP client.
+
+## Claude Code setup
+
+Add the server to your MCP config (see [Claude Code MCP docs][cc-mcp]):
+
+```json
+{
+  "mcpServers": {
+    "VoIP": {
+      "type": "stdio",
+      "command": "uvx",
+      "args": [
+        "-y",
+        "voip[mcp]",
+        "mcp"
+      ],
+      "env": {
+        "SIP_AOR": "sip:****:****@example.com:5060?transport=tcp"
+      }
+    }
+  }
+}
+```
+
+Set `SIP_AOR` to your SIP address-of-record.
+
+## Tools
+
+::: voip.mcp.say
+
+::: voip.mcp.call
+
+[cc-mcp]: https://docs.anthropic.com/en/docs/claude-code/mcp
+[mcp]: https://modelcontextprotocol.io/
diff --git a/mkdocs.yml b/mkdocs.yml
index 082f4bd..1053326 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -6,6 +6,7 @@ watch:
 nav:
   - Usage:
       - Quickstart: index.md
+      - MCP Server: mcp.md
       - Cookbook: cookbook.md
       - Sessions: sessions.md
       - SIP: sip.md
diff --git a/pyproject.toml b/pyproject.toml
index c79b380..a810731 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -41,6 +41,7 @@ dependencies = ["cryptography"]
 audio = ["numpy"]
 hd-audio = ["numpy", "av"]
 ai = ["faster-whisper", "numpy", "av", "ollama", "pocket-tts"]
+mcp = ["fastmcp>=3.2.4", "faster-whisper", "numpy", "av", "ollama", "pocket-tts"]
 cli = ["click", "pygments", "faster-whisper", "numpy", "ollama", "pocket-tts"]
 pygments = ["Pygments"]
 
diff --git a/tests/test_mcp.py b/tests/test_mcp.py
new file mode 100644
index 0000000..0294297
--- /dev/null
+++ b/tests/test_mcp.py
@@ -0,0 +1,489 @@
+"""Tests for the MCP server (voip.mcp)."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+fastmcp = pytest.importorskip("fastmcp")
+pytest.importorskip("faster_whisper")
+pytest.importorskip("pocket_tts")
+
+
+import voip.mcp  # noqa: E402
+from voip.mcp import MCPAgentCall, call, connection_pool, run, say  # noqa: E402
+from voip.rtp import RealtimeTransportProtocol  # noqa: E402
+from voip.sdp.types import MediaDescription, RTPPayloadFormat  # noqa: E402
+from voip.sip.dialog import Dialog  # noqa: E402
+from voip.sip.protocol import SessionInitiationProtocol  # noqa: E402
+from voip.sip.types import CallerID, SipURI  # noqa: E402
+from voip.types import NetworkAddress  # noqa: E402
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def make_media() -> MediaDescription:
+    """Return a minimal MediaDescription for testing."""
+    return MediaDescription(
+        media="audio", port=0, proto="RTP/AVP", fmt=[RTPPayloadFormat(payload_type=8)]
+    )
+
+
+def make_mock_context(reply: str = "Hello!") -> MagicMock:
+    """Return a mock FastMCP Context whose sample() returns *reply*."""
+    ctx = MagicMock(spec=fastmcp.Context)
+    result = MagicMock()
+    result.text = reply
+    ctx.sample = AsyncMock(return_value=result)
+    return ctx
+
+
+def make_agent_call(
+    ctx: MagicMock | None = None,
+    system_prompt: str | None = None,
+    salutation: str = "",
+    messages: list[dict] | None = None,
+) -> MCPAgentCall:
+    """Instantiate a MCPAgentCall with mocked heavy dependencies.
+
+    Patches model-loading factories so no real models are downloaded.
+    """
+    if ctx is None:
+        ctx = make_mock_context()
+    mock_tts = MagicMock()
+    mock_tts.get_state_for_audio_prompt.return_value = {}
+    mock_stt = MagicMock()
+
+    kwargs: dict = {
+        "rtp": MagicMock(spec=RealtimeTransportProtocol),
+        "dialog": MagicMock(spec=Dialog),
+        "media": make_media(),
+        "caller": CallerID(""),
+        "tts_model": mock_tts,
+        "stt_model": mock_stt,
+        "ctx": ctx,
+        "salutation": salutation,
+    }
+    if system_prompt is not None:
+        kwargs["system_prompt"] = system_prompt
+
+    agent = MCPAgentCall(**kwargs)
+    if messages is not None:
+        # Inject synthetic _messages (bypasses AgentCall's auto-population).
+        object.__setattr__(agent, "_messages", messages)
+    return agent
+
+
+# ---------------------------------------------------------------------------
+# MCPAgentCall.transcript
+# ---------------------------------------------------------------------------
+
+
+class TestTranscript:
+    def test_transcript__empty(self) -> None:
+        """Return empty string when only a system message exists."""
+        agent = make_agent_call()
+        assert agent.transcript == ""
+
+    def test_transcript__user_and_assistant(self) -> None:
+        """Format user/assistant turns; exclude system message."""
+        messages = [
+            {"role": "system", "content": "You are helpful."},
+            {"role": "user", "content": "Hi"},
+            {"role": "assistant", "content": "Hello!"},
+        ]
+        agent = make_agent_call(messages=messages)
+        assert agent.transcript == "Caller: Hi\nAgent: Hello!"
+
+    def test_transcript__skips_system_messages(self) -> None:
+        """System messages are excluded from the transcript."""
+        messages = [
+            {"role": "system", "content": "secret"},
+            {"role": "user", "content": "test"},
+        ]
+        agent = make_agent_call(messages=messages)
+        assert "secret" not in agent.transcript
+        assert "Caller: test" in agent.transcript
+
+
+# ---------------------------------------------------------------------------
+# MCPAgentCall.transcription_received
+# ---------------------------------------------------------------------------
+
+
+class TestTranscriptionReceived:
+    async def test_transcription_received__appends_user_message(self) -> None:
+        """Incoming transcription is appended as a user message."""
+        agent = make_agent_call()
+        with patch.object(agent, "cancel_outbound_audio"):
+            with patch("asyncio.create_task"):
+                agent.transcription_received("How are you?")
+
+        user_msgs = [m for m in agent._messages if m["role"] == "user"]
+        assert any(m["content"] == "How are you?" for m in user_msgs)
+
+    async def test_transcription_received__cancels_pending_task(self) -> None:
+        """A pending response task is cancelled before scheduling a new one."""
+        agent = make_agent_call()
+        old_task = MagicMock(spec=asyncio.Task)
+        old_task.done.return_value = False
+        object.__setattr__(agent, "_response_task", old_task)
+
+        with patch.object(agent, "cancel_outbound_audio"):
+            with patch("asyncio.create_task") as mock_create:
+                mock_create.return_value = MagicMock(spec=asyncio.Task)
+                agent.transcription_received("test")
+
+        old_task.cancel.assert_called_once()
+
+    async def test_transcription_received__skips_cancel_when_done(self) -> None:
+        """A completed response task is not cancelled."""
+        agent = make_agent_call()
+        old_task = MagicMock(spec=asyncio.Task)
+        old_task.done.return_value = True
+        object.__setattr__(agent, "_response_task", old_task)
+
+        with patch.object(agent, "cancel_outbound_audio"):
+            with patch("asyncio.create_task"):
+                agent.transcription_received("test")
+
+        old_task.cancel.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# MCPAgentCall.respond
+# ---------------------------------------------------------------------------
+
+
+class TestRespond:
+    async def test_respond__speaks_reply(self) -> None:
+        """respond() speaks the LLM reply and appends it to _messages."""
+        ctx = make_mock_context("Nice to meet you.")
+        agent = make_agent_call(
+            ctx=ctx,
+            messages=[
+                {"role": "system", "content": "Be helpful."},
+                {"role": "user", "content": "Hi"},
+            ],
+        )
+        with patch.object(agent, "send_speech", new_callable=AsyncMock) as mock_send:
+            await agent.respond()
+
+        mock_send.assert_awaited_once_with("Nice to meet you.")
+        assert {"role": "assistant", "content": "Nice to meet you."} in agent._messages
+
+    async def test_respond__filters_system_from_sampling(self) -> None:
+        """System messages are not forwarded to ctx.sample."""
+        ctx = make_mock_context("OK")
+        agent = make_agent_call(
+            ctx=ctx,
+            messages=[
+                {"role": "system", "content": "secret"},
+                {"role": "user", "content": "hello"},
+            ],
+        )
+        with patch.object(agent, "send_speech", new_callable=AsyncMock):
+            await agent.respond()
+
+        args, kwargs = ctx.sample.call_args
+        sent_messages = args[0] if args else kwargs.get("messages", [])
+        roles = [m.role for m in sent_messages]
+        assert "system" not in roles
+
+    async def test_respond__uses_system_prompt_kwarg(self) -> None:
+        """system_prompt is passed as a keyword arg to ctx.sample."""
+        ctx = make_mock_context("OK")
+        agent = make_agent_call(ctx=ctx, system_prompt="Act as an assistant.")
+        with patch.object(agent, "send_speech", new_callable=AsyncMock):
+            await agent.respond()
+
+        _, kwargs = ctx.sample.call_args
+        assert kwargs.get("system_prompt") == "Act as an assistant."
+
+    async def test_respond__empty_reply_is_silent(self) -> None:
+        """An empty or whitespace-only reply does not call send_speech."""
+        ctx = make_mock_context("   ")
+        agent = make_agent_call(
+            ctx=ctx,
+            messages=[
+                {"role": "user", "content": "hello"},
+            ],
+        )
+        with patch.object(agent, "send_speech", new_callable=AsyncMock) as mock_send:
+            await agent.respond()
+
+        mock_send.assert_not_awaited()
+
+    async def test_respond__none_text_is_silent(self) -> None:
+        """A None result.text does not call send_speech."""
+        ctx = MagicMock(spec=fastmcp.Context)
+        result = MagicMock()
+        result.text = None
+        ctx.sample = AsyncMock(return_value=result)
+        agent = make_agent_call(ctx=ctx, messages=[{"role": "user", "content": "hi"}])
+        with patch.object(agent, "send_speech", new_callable=AsyncMock) as mock_send:
+            await agent.respond()
+
+        mock_send.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# say tool
+# ---------------------------------------------------------------------------
+
+
+class TestSayTool:
+    async def test_say__dials_with_parsed_uri(self) -> None:
+        """say() resolves the target via parse_uri relative to the AOR."""
+        aor = SipURI.parse("sip:alice@carrier.example;transport=TLS")
+        mock_sip = MagicMock(spec=SessionInitiationProtocol)
+        mock_sip.aor = aor
+        connection_pool.sip = mock_sip
+
+        ctx = make_mock_context()
+        target_uri = SipURI.parse("sip:bob@carrier.example")
+
+        with patch("voip.mcp.parse_uri", return_value=target_uri) as mock_parse:
+            with patch("voip.mcp.Dialog") as MockDialog:
+                mock_dialog = MagicMock(spec=Dialog)
+                MockDialog.return_value = mock_dialog
+                mock_dialog.dial = AsyncMock()
+
+                await say(ctx=ctx, target="sip:bob@carrier.example", prompt="Hello!")
+
+        mock_parse.assert_called_once_with("sip:bob@carrier.example", aor)
+        MockDialog.assert_called_once_with(sip=mock_sip)
+        mock_dialog.dial.assert_awaited_once()
+        _, kwargs = mock_dialog.dial.call_args
+        assert kwargs["session_class"].__name__ == "SayCall"
+        assert kwargs["text"] == "Hello!"
+
+    async def test_say__empty_prompt(self) -> None:
+        """say() passes an empty string when no prompt is given."""
+        aor = SipURI.parse("sip:alice@example.com")
+        mock_sip = MagicMock(spec=SessionInitiationProtocol)
+        mock_sip.aor = aor
+        connection_pool.sip = mock_sip
+
+        ctx = make_mock_context()
+        with patch("voip.mcp.parse_uri", return_value=aor):
+            with patch("voip.mcp.Dialog") as MockDialog:
+                mock_dialog = MagicMock(spec=Dialog)
+                MockDialog.return_value = mock_dialog
+                mock_dialog.dial = AsyncMock()
+                await say(ctx=ctx, target="sip:bob@example.com")
+
+        _, kwargs = mock_dialog.dial.call_args
+        assert kwargs["text"] == ""
+
+    async def test_say__raises_when_not_connected(self) -> None:
+        """say() raises RuntimeError when connection_pool.sip is not set."""
+        if hasattr(connection_pool, "sip"):
+            del connection_pool.sip
+
+        ctx = make_mock_context()
+        with pytest.raises(RuntimeError, match="run()"):
+            await say(ctx=ctx, target="sip:bob@example.com")
+
+
+# ---------------------------------------------------------------------------
+# call tool
+# ---------------------------------------------------------------------------
+
+
+class TestCallTool:
+    async def test_call__raises_when_not_connected(self) -> None:
+        """call() raises RuntimeError when connection_pool.sip is not set."""
+        if hasattr(connection_pool, "sip"):
+            del connection_pool.sip
+
+        ctx = make_mock_context()
+        with pytest.raises(RuntimeError, match="run()"):
+            await call(ctx=ctx, target="sip:bob@example.com")
+
+    async def test_call__returns_transcript(self) -> None:
+        """call() returns dialog.session.transcript after dialing."""
+        aor = SipURI.parse("sip:alice@example.com")
+        mock_sip = MagicMock(spec=SessionInitiationProtocol)
+        mock_sip.aor = aor
+        connection_pool.sip = mock_sip
+
+        ctx = make_mock_context()
+        target_uri = SipURI.parse("sip:bob@example.com")
+
+        mock_session = MagicMock()
+        mock_session.transcript = "Caller: Hi\nAgent: Hello!"
+
+        with patch("voip.mcp.parse_uri", return_value=target_uri):
+            with patch("voip.mcp.Dialog") as MockDialog:
+                mock_dialog = MagicMock(spec=Dialog)
+                mock_dialog.session = mock_session
+                MockDialog.return_value = mock_dialog
+                mock_dialog.dial = AsyncMock()
+
+                result = await call(
+                    ctx=ctx,
+                    target="sip:bob@example.com",
+                    initial_prompt="Hello!",
+                )
+
+        assert result == "Caller: Hi\nAgent: Hello!"
+        _, kwargs = mock_dialog.dial.call_args
+        assert kwargs["session_class"] is MCPAgentCall
+        assert kwargs["ctx"] is ctx
+        assert kwargs["salutation"] == "Hello!"
+        assert "system_prompt" not in kwargs
+
+    async def test_call__passes_system_prompt_when_given(self) -> None:
+        """call() passes system_prompt to MCPAgentCall when explicitly supplied."""
+        aor = SipURI.parse("sip:alice@example.com")
+        mock_sip = MagicMock(spec=SessionInitiationProtocol)
+        mock_sip.aor = aor
+        connection_pool.sip = mock_sip
+
+        ctx = make_mock_context()
+        with patch("voip.mcp.parse_uri", return_value=aor):
+            with patch("voip.mcp.Dialog") as MockDialog:
+                mock_dialog = MagicMock(spec=Dialog)
+                mock_dialog.session = MagicMock(transcript="")
+                MockDialog.return_value = mock_dialog
+                mock_dialog.dial = AsyncMock()
+
+                await call(
+                    ctx=ctx,
+                    target="sip:bob@example.com",
+                    system_prompt="Act as a robot.",
+                )
+
+        _, kwargs = mock_dialog.dial.call_args
+        assert kwargs["system_prompt"] == "Act as a robot."
+
+    async def test_call__default_empty_initial_prompt(self) -> None:
+        """call() passes empty string as salutation when initial_prompt is omitted."""
+        aor = SipURI.parse("sip:alice@example.com")
+        mock_sip = MagicMock(spec=SessionInitiationProtocol)
+        mock_sip.aor = aor
+        connection_pool.sip = mock_sip
+
+        ctx = make_mock_context()
+        with patch("voip.mcp.parse_uri", return_value=aor):
+            with patch("voip.mcp.Dialog") as MockDialog:
+                mock_dialog = MagicMock(spec=Dialog)
+                mock_dialog.session = MagicMock(transcript="")
+                MockDialog.return_value = mock_dialog
+                mock_dialog.dial = AsyncMock()
+
+                await call(ctx=ctx, target="sip:bob@example.com")
+
+        _, kwargs = mock_dialog.dial.call_args
+        assert kwargs["salutation"] == ""
+
+
+# ---------------------------------------------------------------------------
+# run()
+# ---------------------------------------------------------------------------
+
+
+class TestRun:
+    async def test_run__sets_connection_pool_sip(self) -> None:
+        """run() stores the SIP protocol in connection_pool.sip."""
+        aor = SipURI.parse("sip:alice@example.com")
+        mock_protocol = MagicMock(spec=SessionInitiationProtocol)
+
+        fn = MagicMock()
+        with patch.object(
+            SessionInitiationProtocol,
+            "run",
+            new_callable=AsyncMock,
+            return_value=mock_protocol,
+        ):
+            with patch.object(voip.mcp.mcp, "run_async", new_callable=AsyncMock):
+                await run(fn, aor)
+
+        assert connection_pool.sip is mock_protocol
+
+    async def test_run__calls_mcp_run_async_with_transport(self) -> None:
+        """run() forwards the transport argument to mcp.run_async."""
+        aor = SipURI.parse("sip:alice@example.com")
+        mock_protocol = MagicMock(spec=SessionInitiationProtocol)
+
+        with patch.object(
+            SessionInitiationProtocol,
+            "run",
+            new_callable=AsyncMock,
+            return_value=mock_protocol,
+        ):
+            with patch.object(
+                voip.mcp.mcp, "run_async", new_callable=AsyncMock
+            ) as mock_run:
+                await run(lambda: None, aor, transport="stdio")
+
+        mock_run.assert_awaited_once_with(transport="stdio")
+
+    async def test_run__passes_no_verify_tls(self) -> None:
+        """run() forwards no_verify_tls to SessionInitiationProtocol.run."""
+        aor = SipURI.parse("sip:alice@example.com;transport=TLS")
+        mock_protocol = MagicMock(spec=SessionInitiationProtocol)
+
+        with patch.object(
+            SessionInitiationProtocol,
+            "run",
+            new_callable=AsyncMock,
+            return_value=mock_protocol,
+        ) as mock_sip_run:
+            with patch.object(voip.mcp.mcp, "run_async", new_callable=AsyncMock):
+                await run(lambda: None, aor, no_verify_tls=True)
+
+        _, kwargs = mock_sip_run.call_args
+        assert kwargs["no_verify_tls"] is True
+
+    async def test_run__passes_stun_server(self) -> None:
+        """run() forwards a custom stun_server to SessionInitiationProtocol.run."""
+        aor = SipURI.parse("sip:alice@example.com")
+        stun = NetworkAddress.parse("stun.example.com:3478")
+        mock_protocol = MagicMock(spec=SessionInitiationProtocol)
+
+        with patch.object(
+            SessionInitiationProtocol,
+            "run",
+            new_callable=AsyncMock,
+            return_value=mock_protocol,
+        ) as mock_sip_run:
+            with patch.object(voip.mcp.mcp, "run_async", new_callable=AsyncMock):
+                await run(lambda: None, aor, stun_server=stun)
+
+        _, kwargs = mock_sip_run.call_args
+        assert kwargs["stun_server"] is stun
+
+
+# ---------------------------------------------------------------------------
+# SessionInitiationProtocol.registered_event
+# ---------------------------------------------------------------------------
+
+
+class TestRegisteredEvent:
+    def test_registered_event__set_by_on_registered(self) -> None:
+        """on_registered() sets registered_event so run() can unblock."""
+        protocol = SessionInitiationProtocol.__new__(SessionInitiationProtocol)
+        protocol.registered_event = asyncio.Event()
+        protocol.ready_callback = None
+
+        assert not protocol.registered_event.is_set()
+        protocol.on_registered()
+        assert protocol.registered_event.is_set()
+
+    def test_registered_event__ready_callback_called_after_event(self) -> None:
+        """ready_callback is invoked after registered_event is set."""
+        call_order: list[str] = []
+        protocol = SessionInitiationProtocol.__new__(SessionInitiationProtocol)
+        protocol.registered_event = asyncio.Event()
+
+        def _cb() -> None:
+            call_order.append("cb" if protocol.registered_event.is_set() else "early")
+
+        protocol.ready_callback = _cb
+        protocol.on_registered()
+
+        assert call_order == ["cb"]
diff --git a/voip/__main__.py b/voip/__main__.py
index e68207a..7813899 100644
--- a/voip/__main__.py
+++ b/voip/__main__.py
@@ -31,6 +31,14 @@
 logger = logging.getLogger("voip")
 
 
+def _parse_sip_uri(ctx, param, value) -> SipURI:
+    """Parse a SIP URI."""
+    try:
+        return SipURI.parse(value)
+    except ValueError as e:
+        raise click.BadParameter(str(e)) from e
+
+
 @dataclasses.dataclass(kw_only=True, slots=True)
 class ConsoleMessageProtocol(SessionInitiationProtocol):
     """Pretty print SIP messages to stdout using pygments."""
@@ -93,8 +101,56 @@ def voip(ctx, verbose: int = 0):
     logging.getLogger("voip").setLevel(max(10, 10 * (3 - verbose)))
 
 
+@voip.command()
+@click.argument(
+    "aor",
+    metavar="AOR",
+    envvar="SIP_AOR",
+    callback=_parse_sip_uri,
+)
+@click.option(
+    "--stun-server",
+    envvar="STUN_SERVER",
+    default="stun.cloudflare.com:3478",
+    show_default=True,
+    metavar="HOST[:PORT]",
+    callback=lambda ctx, param, value: NetworkAddress.parse(value),
+    is_eager=False,
+    help="STUN server for RTP NAT traversal.",
+)
+@click.option(
+    "--no-verify-tls",
+    is_flag=True,
+    default=False,
+    help="Disable TLS certificate verification (insecure; for testing only).",
+)
+@click.option(
+    "--transport",
+    type=click.Choice(["http", "stdio"]),
+    default="stdio",
+    show_default=True,
+)
+def mcp(aor: SipURI, stun_server: NetworkAddress, no_verify_tls: bool, transport: str):
+    from .mcp import run
+
+    asyncio.run(
+        run(
+            lambda: None,
+            aor,
+            stun_server=stun_server,
+            no_verify_tls=no_verify_tls,
+            transport=transport,
+        )
+    )
+
+
 @voip.group()
-@click.argument("aor", metavar="AOR", envvar="SIP_AOR")
+@click.argument(
+    "aor",
+    metavar="AOR",
+    envvar="SIP_AOR",
+    callback=_parse_sip_uri,
+)
 @click.option(
     "--stun-server",
     envvar="STUN_SERVER",
@@ -115,14 +171,9 @@ def voip(ctx, verbose: int = 0):
 def sip(ctx, aor, stun_server, no_verify_tls):
     """Session Initiation Protocol (SIP)."""
     ctx.ensure_object(dict)
-    try:
-        parsed_aor = SipURI.parse(aor)
-    except ValueError as exc:
-        raise click.BadParameter(str(exc), param_hint="AOR") from exc
-
     ctx.obj.update(
-        aor=parsed_aor,
-        proxy_addr=parsed_aor.maddr,
+        aor=aor,
+        proxy_addr=aor.maddr,
         stun_server=stun_server,
         no_verify_tls=no_verify_tls,
     )
diff --git a/voip/mcp.py b/voip/mcp.py
new file mode 100644
index 0000000..17abbeb
--- /dev/null
+++ b/voip/mcp.py
@@ -0,0 +1,158 @@
+"""MCP server for VoIP actions.
+
+Requires the ``mcp`` extra: ``pip install voip[mcp]``.
+"""
+
+import asyncio
+import dataclasses
+import threading
+import typing
+
+from fastmcp import Context, FastMCP
+from mcp.types import SamplingMessage, TextContent
+
+import voip
+from voip import ai
+from voip.ai import SayCall
+from voip.sip import Dialog
+from voip.sip.protocol import SessionInitiationProtocol
+from voip.sip.types import SipURI, parse_uri
+from voip.types import NetworkAddress
+
+__all__ = [
+    "mcp",
+    "run",
+    "MCPAgentCall",
+]
+
+mcp = FastMCP(
+    "VoIP",
+    "Provide a set of tools to make phone calls.",
+    version=voip.__version__,
+    website_url="https://codingjoe.dev/VoIP/",
+)
+
+#: Thread-local storage holding the active [`SessionInitiationProtocol`][voip.sip.protocol.SessionInitiationProtocol].
+#: Populated by [`run`][voip.mcp.run] before the MCP server starts.
+connection_pool = threading.local()
+
+
+@dataclasses.dataclass(kw_only=True, slots=True)
+class MCPAgentCall(ai.AgentCall):
+    """
+    Agent call that generates voice responses via MCP sampling.
+
+    Replaces the Ollama backend of [`AgentCall`][voip.ai.AgentCall] with the
+    MCP client's language model via MCP's sampling API.
+    """
+
+    ctx: Context
+
+    @property
+    def transcript(self) -> str:
+        return "\n".join(
+            f"{'Caller' if msg['role'] == 'user' else 'Agent'}: {msg['content']}"
+            for msg in self._messages
+            if msg["role"] != "system"
+        )
+
+    def transcription_received(self, text: str) -> None:
+        self.cancel_outbound_audio()
+        self._messages.append({"role": "user", "content": text})
+        if self._response_task is not None and not self._response_task.done():
+            self._response_task.cancel()
+        self._response_task = asyncio.create_task(self.respond())
+
+    async def respond(self) -> None:
+        sampling_messages = [
+            SamplingMessage(
+                role=typing.cast(typing.Literal["user", "assistant"], msg["role"]),
+                content=TextContent(type="text", text=msg["content"]),
+            )
+            for msg in self._messages
+            if msg["role"] != "system"
+        ]
+        result = await self.ctx.sample(
+            sampling_messages,
+            system_prompt=self.system_prompt,
+        )
+        if result.text and (reply := result.text.strip()):
+            self._messages.append({"role": "assistant", "content": reply})
+            await self.send_speech(reply)
+
+
+@mcp.tool
+async def say(ctx: Context, target: str, prompt: str = "") -> None:
+    """Call a phone number and speak a message.
+
+    Dials `target`, synthesises `prompt` as speech, then hangs
+    up automatically once the message has been delivered.
+
+    Args:
+        ctx: FastMCP context (injected automatically by the framework).
+        target: Phone number or SIP URI to call, e.g. `"tel:+1234567890"`
+            or `"sip:alice@example.com"`.
+        prompt: Text to speak during the call.
+    """
+    if not hasattr(connection_pool, "sip"):
+        raise RuntimeError("VoIP not connected: call run() before using tools.")
+    target_uri = parse_uri(target, connection_pool.sip.aor)
+    dialog = Dialog(sip=connection_pool.sip)
+    await dialog.dial(target_uri, session_class=SayCall, text=prompt)
+
+
+@mcp.tool
+async def call(
+    ctx: Context,
+    target: str,
+    initial_prompt: str = "",
+    system_prompt: str | None = None,
+) -> str:
+    """Call a phone number, hold a conversation, and return the transcript.
+
+    Dials `target`, optionally speaks `initial_prompt`, then drives a conversation.
+
+    Returns once the remote party hangs up.
+
+    Args:
+        target: Phone number or SIP URI to call, e.g. `"tel:+1234567890"`
+            or `"sip:alice@example.com"`.
+        initial_prompt: Opening message spoken when the call connects.
+            Pass an empty string to suppress the default greeting.
+        system_prompt: System instruction passed to the language model.
+            Defaults to [`AgentCall.system_prompt`][voip.ai.AgentCall].
+
+    Returns:
+        The full conversation transcript with `Caller:` / `Agent:` prefixes.
+    """
+    if not hasattr(connection_pool, "sip"):
+        raise RuntimeError("VoIP not connected: call run() before using tools.")
+    target_uri = parse_uri(target, connection_pool.sip.aor)
+    dialog = Dialog(sip=connection_pool.sip)
+    kwargs: dict[str, typing.Any] = {"ctx": ctx, "salutation": initial_prompt}
+    if system_prompt is not None:
+        kwargs["system_prompt"] = system_prompt
+    await dialog.dial(target_uri, session_class=MCPAgentCall, **kwargs)
+    return dialog.session.transcript
+
+
+async def run(
+    fn: typing.Callable[[], None],
+    aor: SipURI,
+    *,
+    no_verify_tls: bool = False,
+    stun_server: NetworkAddress | None = None,
+    transport: str | None = None,
+) -> None:
+    connection_pool.sip = await SessionInitiationProtocol.run(
+        fn,
+        aor,
+        Dialog,
+        no_verify_tls=no_verify_tls,
+        stun_server=stun_server,
+    )
+    await mcp.run_async(transport=transport)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    mcp.run()
diff --git a/voip/sip/dialog.py b/voip/sip/dialog.py
index d8c90c3..5b2ba7c 100644
--- a/voip/sip/dialog.py
+++ b/voip/sip/dialog.py
@@ -73,6 +73,7 @@ class MySession(SessionInitiationProtocol):
         default=None, compare=False, repr=False
     )
 
+    session: Session | None = dataclasses.field(default=None, init=False, compare=False)
     created: datetime.datetime = dataclasses.field(
         init=False, default_factory=datetime.datetime.now
     )
diff --git a/voip/sip/protocol.py b/voip/sip/protocol.py
index c574d9f..ef549e6 100644
--- a/voip/sip/protocol.py
+++ b/voip/sip/protocol.py
@@ -7,7 +7,9 @@
 import asyncio
 import dataclasses
 import datetime
+import ipaddress
 import logging
+import ssl
 import typing
 
 from voip.rtp import RealtimeTransportProtocol
@@ -110,13 +112,86 @@ async def main():
     disconnected_event: asyncio.Event = dataclasses.field(
         init=False, default_factory=asyncio.Event
     )
+    registered_event: asyncio.Event = dataclasses.field(
+        init=False, default_factory=asyncio.Event
+    )
     transport: asyncio.Transport | None = dataclasses.field(init=False, default=None)
     is_secure: bool = dataclasses.field(init=False, default=False)
     recv_buffer: bytearray = dataclasses.field(init=False, default_factory=bytearray)
+    ready_callback: typing.Callable[[], None] | None = dataclasses.field(
+        default=None, repr=False, compare=False
+    )
 
     def __post_init__(self):
         self.public_address = self.public_address or self.rtp.public_address
 
+    @classmethod
+    async def run(
+        cls,
+        fn: typing.Callable[[], None],
+        aor: types.SipURI,
+        dialog_class: type[Dialog],
+        *,
+        no_verify_tls: bool = False,
+        stun_server: NetworkAddress | None = None,
+    ) -> SessionInitiationProtocol:
+        """Run a SIP session and call *fn* once registered.
+
+        Establishes RTP and SIP/TLS connections derived from *aor*, then
+        **suspends until SIP registration is confirmed** before returning the
+        ready protocol.  After this call returns, the MCP server (or any other
+        caller) may safely place outbound calls.
+
+        The transport protocol (TLS vs plain TCP) and proxy address are read
+        from *aor* directly — no extra arguments are needed.
+
+        Args:
+            fn: Called when the SIP session is registered, before
+                `run` returns. Receives no arguments. May use
+                [`asyncio.create_task`][] for async work.
+            aor: SIP Address of Record, e.g. ``sip:alice@carrier.example``.
+                The host, port, and ``transport`` parameter are used to connect
+                to the SIP proxy.
+            dialog_class: [`Dialog`][voip.sip.Dialog] subclass used for
+                inbound calls. Defaults to the base
+                [`Dialog`][voip.sip.Dialog], which rejects all calls.
+            no_verify_tls: Disable TLS certificate verification. Insecure; for
+                testing only. Defaults to ``False``.
+            stun_server: STUN server for RTP NAT traversal. Defaults to
+                ``stun.cloudflare.com:3478``.
+
+        Returns:
+            The registered [`SessionInitiationProtocol`][voip.sip.protocol.SessionInitiationProtocol]
+            instance, ready to place calls.
+        """
+        loop = asyncio.get_running_loop()
+
+        rtp_bind_address = (
+            "::" if isinstance(aor.maddr[0], ipaddress.IPv6Address) else "0.0.0.0"
+        )  # noqa: S104
+        _, rtp_protocol = await loop.create_datagram_endpoint(
+            lambda: RealtimeTransportProtocol(stun_server_address=stun_server),
+            local_addr=(rtp_bind_address, 0),
+        )
+
+        ssl_context: ssl.SSLContext | None = None
+        if aor.transport == "TLS":
+            ssl_context = ssl.create_default_context()
+            if no_verify_tls:
+                ssl_context.check_hostname = False
+                ssl_context.verify_mode = ssl.CERT_NONE
+
+        _, protocol = await loop.create_connection(
+            lambda: cls(
+                aor=aor, rtp=rtp_protocol, dialog_class=dialog_class, ready_callback=fn
+            ),
+            host=str(aor.maddr[0]),
+            port=aor.maddr[1],
+            ssl=ssl_context,
+        )
+        await protocol.registered_event.wait()
+        return protocol
+
     def register_dialog(self, dialog: Dialog) -> None:
         """Register *dialog* keyed by ``(dialog.local_tag, dialog.remote_tag)``."""
         if dialog.remote_tag is None:
@@ -360,6 +435,9 @@ def on_registered(self) -> None:
         Override in subclasses to initiate outbound calls or start other
         post-registration activity. The base implementation is a no-op.
         """
+        self.registered_event.set()
+        if self.ready_callback is not None:
+            self.ready_callback()
 
     @property
     def contact(self) -> str:
diff --git a/voip/sip/transactions.py b/voip/sip/transactions.py
index a941c44..d515dfd 100644
--- a/voip/sip/transactions.py
+++ b/voip/sip/transactions.py
@@ -754,7 +754,7 @@ def _start_call(self, response: Response) -> None:
             )
 
         if self.pending_call_class is not None:
-            call_handler = self.pending_call_class(
+            self.dialog.session = self.pending_call_class(
                 rtp=self.sip.rtp,
                 caller=CallerID(str(self.sip.aor)),
                 media=negotiated_media,
@@ -780,7 +780,7 @@ def _start_call(self, response: Response) -> None:
                 )
             else:
                 remote_rtp_address = None
-            self.sip.rtp.register_call(remote_rtp_address, call_handler)
+            self.sip.rtp.register_call(remote_rtp_address, self.dialog.session)
             if remote_rtp_address is not None:
                 self.sip.rtp.send(b"\x00", remote_rtp_address)
 
diff --git a/voip/stun.py b/voip/stun.py
index 4aab386..e19311e 100644
--- a/voip/stun.py
+++ b/voip/stun.py
@@ -111,7 +111,9 @@ def packet_received(self, data: bytes, addr: tuple[str, int]) -> None:
     ```
     """
 
-    stun_server_address: tuple[str, int] | None = ("stun.cloudflare.com", 3478)
+    stun_server_address: NetworkAddress | None = NetworkAddress(
+        "stun.cloudflare.com", 3478
+    )
     _stun_transaction_id: bytes = dataclasses.field(init=False, default=b"")
     transport: asyncio.DatagramTransport | None = dataclasses.field(
         init=False, default=None
@@ -135,9 +137,9 @@ def stun_connection_made(
     ) -> None:
         """Called when the socket is ready and the reachable address is known.
 
-        When STUN is configured, *addr* is the **public** ``(ip, port)``
+        When STUN is configured, *addr* is the **public** `(ip, port)`
         discovered from the STUN Binding Response and this method is called
-        by `datagram_received`.  When ``stun_server_address=None``,
+        by `datagram_received`.  When `stun_server_address=None`,
         *addr* is the local socket address and this method is called
         synchronously from `connection_made`.
 
@@ -146,7 +148,7 @@ def stun_connection_made(
 
         Args:
             transport: The UDP transport bound to this protocol.
-            addr: Reachable ``(host, port)`` — public when STUN is used,
+            addr: Reachable `(host, port)` — public when STUN is used,
                 local otherwise.
         """  # noqa: D401
 
@@ -155,10 +157,10 @@ def send(self, data: bytes, addr: NetworkAddress) -> None:
 
         Args:
             data: Raw bytes to transmit.
-            addr: Destination ``(host, port)``.
+            addr: Destination `(host, port)`.
         """
         if self.transport is not None:
-            self.transport.sendto(data, addr)
+            self.transport.sendto(data, (str(addr[0]), addr[1]))
 
     def close(self) -> None:
         """Close the underlying UDP transport."""