diff --git a/src/openjarvis/cli/ask.py b/src/openjarvis/cli/ask.py
index ff9f6c15..c0c9ad8a 100644
--- a/src/openjarvis/cli/ask.py
+++ b/src/openjarvis/cli/ask.py
@@ -339,7 +339,9 @@ def _run_agent(
 
     agent_cls = AgentRegistry.get(agent_name)
 
-    # Build tools
+    # Build tools — local registry tools + MCP server tools from config
+    # (#461 — MCP tools were silently dropped because ask.py only used
+    # ToolRegistry).
     tools = []
     if tool_names:
         # Trigger tool registration
@@ -347,6 +349,26 @@ def _run_agent(
 
         tools = _build_tools(tool_names, config, engine, model_name)
 
+    # MCP tools from config.tools.mcp.servers. Loaded regardless of
+    # tool_names — if the caller passed --tools, the loader filters MCP
+    # tools to those names; otherwise every MCP tool is included.
+    from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+    mcp_tools, mcp_clients = load_mcp_tools_from_config(
+        config.tools.mcp,
+        allowed_names=set(tool_names) if tool_names else None,
+    )
+    if mcp_tools:
+        # Dedup against registry tools by spec.name — first occurrence wins
+        # so a registry tool always takes precedence over an MCP tool with
+        # the same name (avoids the "two tools with the same name" footgun
+        # the verdict flagged).
+        existing = {t.spec.name for t in tools}
+        for t in mcp_tools:
+            if t.spec.name not in existing:
+                tools.append(t)
+                existing.add(t.spec.name)
+
     # Build agent with appropriate kwargs
     agent_kwargs = {
         "bus": bus,
@@ -378,6 +400,12 @@ def _run_agent(
         )
 
     agent = agent_cls(engine, model_name, **agent_kwargs)
+    # Hold MCP transports alive for the agent's lifetime — without this
+    # reference they'd be garbage-collected when this function returns
+    # and the underlying HTTP connections would close mid-execution (#461
+    # adversarial review caught this).
+    if mcp_clients:
+        agent._mcp_clients = mcp_clients
     ctx = AgentContext()
 
     # Inject memory context into conversation if available
diff --git a/src/openjarvis/cli/serve.py b/src/openjarvis/cli/serve.py
index abf8815c..29f7fdf3 100644
--- a/src/openjarvis/cli/serve.py
+++ b/src/openjarvis/cli/serve.py
@@ -188,6 +188,11 @@ def serve(
                 if sec.capability_policy is not None:
                     agent_kwargs["capability_policy"] = sec.capability_policy
 
+                # MCP transports persisted on the agent at the bottom of
+                # this block — initialise here so the reference is valid
+                # even when accepts_tools is False (#461).
+                mcp_clients: list = []
+
                 # Load tools for agents that support them
                 if getattr(agent_cls, "accepts_tools", False):
                     import openjarvis.tools  # noqa: F401  # trigger registration
@@ -221,6 +226,22 @@ def serve(
                             tools.append(tool_cls())
                         elif isinstance(tool_cls, BaseTool):
                             tools.append(tool_cls)
+
+                    # MCP server tools from config.tools.mcp.servers
+                    # (#461 — these were silently dropped).
+                    from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+                    mcp_tools, mcp_clients = load_mcp_tools_from_config(
+                        config.tools.mcp,
+                        allowed_names=allowed if configured else None,
+                    )
+                    if mcp_tools:
+                        existing = {t.spec.name for t in tools}
+                        for t in mcp_tools:
+                            if t.spec.name not in existing:
+                                tools.append(t)
+                                existing.add(t.spec.name)
+
                     if tools:
                         agent_kwargs["tools"] = tools
 
@@ -228,6 +249,10 @@ def serve(
                     agent_kwargs["max_turns"] = config.agent.max_turns
 
                 agent = agent_cls(engine, model_name, **agent_kwargs)
+                # Pin MCP transports to the agent's lifetime so HTTP
+                # connections don't close mid-request (#461).
+                if mcp_clients:
+                    agent._mcp_clients = mcp_clients
         except Exception as exc:
             import traceback
 
@@ -260,6 +285,10 @@ def serve(
         channel_agent = config.channel.default_agent or agent_key or "simple"
 
         _channel_tools: list = []
+        # MCP transports persisted at function scope (= server-process
+        # lifetime); see the comment near the channel-MCP-load block
+        # below. Initialise here so it's always bound. #461.
+        _channel_mcp_clients: list = []
         if channel_agent:
             try:
                 import openjarvis.agents
@@ -298,8 +327,30 @@ def serve(
                                 _channel_tools.append(_tcls())
                             elif isinstance(_tcls, BaseTool):
                                 _channel_tools.append(_tcls)
+
+                        # MCP tools for the channel agent too (#461).
+                        from openjarvis.mcp.loader import (
+                            load_mcp_tools_from_config,
+                        )
+
+                        _ch_mcp_tools, _ch_mcp_clients = load_mcp_tools_from_config(
+                            config.tools.mcp,
+                            allowed_names=_allowed if configured else None,
+                        )
+                        if _ch_mcp_tools:
+                            _existing = {t.spec.name for t in _channel_tools}
+                            for t in _ch_mcp_tools:
+                                if t.spec.name not in _existing:
+                                    _channel_tools.append(t)
+                                    _existing.add(t.spec.name)
+                        # Hold a reference at module / function scope —
+                        # the channel agent is constructed inside
+                        # JarvisSystem below; we extend its lifetime by
+                        # keeping the list bound here.
+                        _channel_mcp_clients = _ch_mcp_clients
             except Exception as exc:
                 logger.warning("Channel tools failed to load: %s", exc)
+                _channel_mcp_clients = []
 
         _wire_system = JarvisSystem(
             config=config,
diff --git a/src/openjarvis/mcp/loader.py b/src/openjarvis/mcp/loader.py
new file mode 100644
index 00000000..b70a6277
--- /dev/null
+++ b/src/openjarvis/mcp/loader.py
@@ -0,0 +1,133 @@
+"""Shared helper for loading MCP server tools from a TOML config blob.
+
+Used by ``cli/ask.py``, ``cli/serve.py``, ``system/builder.py`` and
+``server/agent_manager_routes.py`` so each call site doesn't reimplement
+the server-config → transport → client → discovered-tools pipeline.
+
+The returned tuple of ``(tools, clients)`` is load-bearing: the caller
+MUST hold a reference to ``clients`` for as long as the tools are used,
+otherwise the MCP transport sessions get garbage-collected and the
+underlying HTTP connections close mid-execution (see #461 adversarial
+review). The recommended pattern is to stash the client list on the
+agent so they share its lifetime:
+
+    tools, mcp_clients = load_mcp_tools_from_config(config.tools.mcp)
+    agent = AgentCls(tools=tools, ...)
+    agent._mcp_clients = mcp_clients   # keep transports alive
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import TYPE_CHECKING, Any, Optional
+
+if TYPE_CHECKING:
+    from openjarvis.core.types import ToolSpec  # noqa: F401
+    from openjarvis.mcp.client import MCPClient
+    from openjarvis.tools._stubs import BaseTool
+
+logger = logging.getLogger(__name__)
+
+
+def load_mcp_tools_from_config(
+    mcp_cfg: Any,
+    *,
+    allowed_names: Optional[set[str]] = None,
+) -> tuple[list["BaseTool"], list["MCPClient"]]:
+    """Load tools from every server in ``mcp_cfg.servers``.
+
+    Returns ``(tools, clients)``. ``clients`` is the list of live
+    ``MCPClient`` instances — keep a reference or the transports get
+    GC'd. Failures in any single server are logged and that server is
+    skipped; the rest are returned as a best-effort batch.
+
+    ``allowed_names`` is an outer filter applied after each server's
+    own include/exclude filter. Pass the caller's `--tools`/`enabled`
+    list to honour CLI scoping; pass ``None`` to take every tool.
+
+    Returns ``([], [])`` when mcp is disabled or no servers are
+    configured — no exception, no warning.
+    """
+    # ``enabled`` and ``servers`` come from openjarvis.core.config's
+    # MCPConfig dataclass; accept duck-typed equivalents for tests.
+    enabled = getattr(mcp_cfg, "enabled", False)
+    servers_blob = getattr(mcp_cfg, "servers", None)
+    if not enabled or not servers_blob:
+        return [], []
+
+    try:
+        server_list = (
+            json.loads(servers_blob) if isinstance(servers_blob, str) else servers_blob
+        )
+    except (json.JSONDecodeError, TypeError) as exc:
+        logger.warning("Failed to parse MCP servers config: %s", exc)
+        return [], []
+    if not isinstance(server_list, list):
+        logger.warning(
+            "MCP servers config is not a list (got %r) — skipping MCP discovery",
+            type(server_list).__name__,
+        )
+        return [], []
+
+    # Imported lazily so that `openjarvis.mcp.loader` can be imported
+    # cheaply from CLI startup paths without dragging in the heavy MCP
+    # client stack until something actually wants to discover tools.
+    from openjarvis.mcp.client import MCPClient
+    from openjarvis.mcp.transport import StdioTransport, StreamableHTTPTransport
+    from openjarvis.tools.mcp_adapter import MCPToolProvider
+
+    tools: list["BaseTool"] = []
+    clients: list["MCPClient"] = []
+
+    for server_cfg in server_list:
+        try:
+            cfg = (
+                json.loads(server_cfg) if isinstance(server_cfg, str) else server_cfg
+            )
+            name = cfg.get("name", "<unnamed>")
+            url = cfg.get("url")
+            token = cfg.get("token")
+            command = cfg.get("command", "")
+            args = cfg.get("args", [])
+
+            if url:
+                transport = StreamableHTTPTransport(url=url, token=token)
+            elif command:
+                transport = StdioTransport(command=[command] + args)
+            else:
+                logger.warning(
+                    "MCP server '%s' has neither 'url' nor 'command' — skipping",
+                    name,
+                )
+                continue
+
+            client = MCPClient(transport)
+            client.initialize()
+            clients.append(client)
+
+            provider = MCPToolProvider(client)
+            discovered = provider.discover()
+
+            include_tools = set(cfg.get("include_tools", []))
+            exclude_tools = set(cfg.get("exclude_tools", []))
+            if include_tools:
+                discovered = [t for t in discovered if t.spec.name in include_tools]
+            if exclude_tools:
+                discovered = [t for t in discovered if t.spec.name not in exclude_tools]
+            if allowed_names:
+                discovered = [t for t in discovered if t.spec.name in allowed_names]
+
+            tools.extend(discovered)
+            logger.info(
+                "Discovered %d MCP tools from server '%s'", len(discovered), name
+            )
+        except Exception as exc:  # per-server isolation
+            logger.warning(
+                "Failed to discover MCP tools from '%s': %s",
+                cfg.get("name", "<unnamed>") if "cfg" in locals() else "<unparsed>",
+                exc,
+            )
+            continue
+
+    return tools, clients
diff --git a/src/openjarvis/mcp/transport.py b/src/openjarvis/mcp/transport.py
index 11f8020f..fa26d0d2 100644
--- a/src/openjarvis/mcp/transport.py
+++ b/src/openjarvis/mcp/transport.py
@@ -122,12 +122,14 @@ class StreamableHTTPTransport(MCPTransport):
         self,
         url: str,
         *,
+        token: Optional[str] = None,
         connect_timeout: float = 10.0,
         request_timeout: float = 60.0,
     ) -> None:
         import httpx
 
         self._url = url
+        self._token = token
         self._session_id: Optional[str] = None
         self._client = httpx.Client(
             timeout=httpx.Timeout(
@@ -146,11 +148,21 @@ class StreamableHTTPTransport(MCPTransport):
         return f"{parsed.scheme}://{parsed.netloc}"
 
     def _build_headers(self) -> dict:
-        """Build common request headers."""
+        """Build common request headers.
+
+        Sends ``Authorization: Bearer <token>`` when the transport was
+        constructed with a token (#461) — required by authenticated MCP
+        servers such as Home Assistant's. Falsy tokens (None / empty
+        string) deliberately do NOT send the header, matching the
+        upstream MCP spec and the cfg.get("token") plumbing in the
+        builder.
+        """
         headers = {
             "Content-Type": "application/json",
             "Accept": "application/json, text/event-stream",
         }
+        if self._token:
+            headers["Authorization"] = f"Bearer {self._token}"
         if self._session_id is not None:
             headers["Mcp-Session-Id"] = self._session_id
         return headers
diff --git a/src/openjarvis/server/agent_manager_routes.py b/src/openjarvis/server/agent_manager_routes.py
index 9c2da535..cd6061a3 100644
--- a/src/openjarvis/server/agent_manager_routes.py
+++ b/src/openjarvis/server/agent_manager_routes.py
@@ -687,12 +687,14 @@ def _get_mcp_tools(app_state: Any) -> Tuple[List[Dict[str, Any]], Dict[str, Any]
         cfg = _json.loads(server_cfg) if isinstance(server_cfg, str) else server_cfg
         name = cfg.get("name", "<unnamed>")
         url = cfg.get("url")
+        # Bearer token from config — mirrors the builder.py fix for #461.
+        token = cfg.get("token")
         command = cfg.get("command", "")
         args = cfg.get("args", [])
 
         try:
             if url:
-                transport = StreamableHTTPTransport(url=url)
+                transport = StreamableHTTPTransport(url=url, token=token)
             elif command:
                 transport = StdioTransport(command=[command] + args)
             else:
diff --git a/src/openjarvis/system/builder.py b/src/openjarvis/system/builder.py
index 1197ade7..a5a317c3 100644
--- a/src/openjarvis/system/builder.py
+++ b/src/openjarvis/system/builder.py
@@ -584,11 +584,14 @@ class SystemBuilder:
         cfg = json.loads(server_cfg) if isinstance(server_cfg, str) else server_cfg
         name = cfg.get("name", "<unnamed>")
         url = cfg.get("url")
+        # Bearer token from config — needed by authenticated MCP servers
+        # like Home Assistant. None / empty string skips the header. #461.
+        token = cfg.get("token")
         command = cfg.get("command", "")
         args = cfg.get("args", [])
 
         if url:
-            transport = StreamableHTTPTransport(url=url)
+            transport = StreamableHTTPTransport(url=url, token=token)
         elif command:
             transport = StdioTransport(command=[command] + args)
         else:
diff --git a/tests/mcp/test_discovery.py b/tests/mcp/test_discovery.py
index 6e516966..1bfde7cf 100644
--- a/tests/mcp/test_discovery.py
+++ b/tests/mcp/test_discovery.py
@@ -53,7 +53,11 @@ class TestDiscoverHTTPServer:
         cfg = {"name": "ha-mcp", "url": "http://172.16.3.1:9583/mcp"}
         result = builder._discover_external_mcp(cfg)
 
-        mock_transport_cls.assert_called_once_with(url="http://172.16.3.1:9583/mcp")
+        # token=None is now forwarded explicitly (#461) so authenticated
+        # MCP servers can use it; missing config field → None → no header.
+        mock_transport_cls.assert_called_once_with(
+            url="http://172.16.3.1:9583/mcp", token=None
+        )
         mock_client_cls.return_value.initialize.assert_called_once()
         assert len(result) == 2
         assert result[0].spec.name == "get_entities"
@@ -192,4 +196,8 @@ class TestStringConfig:
         cfg_str = json.dumps({"name": "test", "url": "http://localhost:8080/mcp"})
         builder._discover_external_mcp(cfg_str)
 
-        mock_transport_cls.assert_called_once_with(url="http://localhost:8080/mcp")
+        # token=None is forwarded by the builder (#461) — see comment in
+        # TestDiscoverHTTPServer.test_url_config_uses_http_transport.
+        mock_transport_cls.assert_called_once_with(
+            url="http://localhost:8080/mcp", token=None
+        )
diff --git a/tests/mcp/test_loader.py b/tests/mcp/test_loader.py
new file mode 100644
index 00000000..e939dd89
--- /dev/null
+++ b/tests/mcp/test_loader.py
@@ -0,0 +1,235 @@
+"""Regression tests for openjarvis.mcp.loader.load_mcp_tools_from_config.
+
+Closes the gap that #461 surfaced — MCP tools were silently dropped on
+`jarvis ask` and `jarvis serve` because neither path read
+`config.tools.mcp.servers`. The loader is the shared helper they now
+both call.
+"""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def _make_mcp_cfg(*, enabled=True, servers):
+    """Build a duck-typed MCPConfig with enabled flag + servers JSON."""
+    cfg = MagicMock()
+    cfg.enabled = enabled
+    cfg.servers = (
+        json.dumps(servers) if not isinstance(servers, str) else servers
+    )
+    return cfg
+
+
+def _fake_tool(name):
+    t = MagicMock()
+    t.spec.name = name
+    return t
+
+
+@pytest.fixture
+def _mock_mcp_stack():
+    """Patch MCPClient / transports / MCPToolProvider so no real I/O happens."""
+    with patch("openjarvis.mcp.client.MCPClient") as MockClient, patch(
+        "openjarvis.mcp.transport.StreamableHTTPTransport"
+    ) as MockHttp, patch(
+        "openjarvis.mcp.transport.StdioTransport"
+    ) as MockStdio, patch(
+        "openjarvis.tools.mcp_adapter.MCPToolProvider"
+    ) as MockProvider:
+        # Default: any provider discovers no tools (per-test overrides as needed)
+        MockProvider.return_value.discover.return_value = []
+        MockClient.return_value.initialize.return_value = None
+        yield {
+            "client": MockClient,
+            "http": MockHttp,
+            "stdio": MockStdio,
+            "provider": MockProvider,
+        }
+
+
+class TestLoaderEarlyReturns:
+    def test_disabled_returns_empty(self, _mock_mcp_stack):
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        cfg = _make_mcp_cfg(enabled=False, servers=[{"url": "http://x"}])
+        tools, clients = load_mcp_tools_from_config(cfg)
+        assert tools == []
+        assert clients == []
+        # Critical: when disabled, we don't even instantiate transports
+        _mock_mcp_stack["http"].assert_not_called()
+
+    def test_empty_servers_returns_empty(self, _mock_mcp_stack):
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        cfg = _make_mcp_cfg(enabled=True, servers=[])
+        tools, clients = load_mcp_tools_from_config(cfg)
+        assert tools == []
+        assert clients == []
+
+    def test_malformed_json_logs_warning_returns_empty(self, _mock_mcp_stack, caplog):
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        cfg = MagicMock()
+        cfg.enabled = True
+        cfg.servers = "{not valid json"
+        with caplog.at_level("WARNING"):
+            tools, clients = load_mcp_tools_from_config(cfg)
+        assert tools == []
+        assert clients == []
+        assert any("parse MCP servers config" in r.message for r in caplog.records)
+
+
+class TestLoaderTokenPlumbing:
+    def test_token_passed_to_streamable_http(self, _mock_mcp_stack):
+        """Regression for #461 — token in cfg → StreamableHTTPTransport(token=...)."""
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        cfg = _make_mcp_cfg(
+            enabled=True,
+            servers=[
+                {
+                    "name": "home-assistant",
+                    "url": "http://homeassistant.local:8123/mcp",
+                    "token": "ha-llat-secret",
+                }
+            ],
+        )
+        load_mcp_tools_from_config(cfg)
+        _mock_mcp_stack["http"].assert_called_once_with(
+            url="http://homeassistant.local:8123/mcp",
+            token="ha-llat-secret",
+        )
+
+    def test_no_token_passes_none(self, _mock_mcp_stack):
+        """Missing token in cfg → token=None (not 'undefined' or KeyError)."""
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        cfg = _make_mcp_cfg(
+            enabled=True,
+            servers=[{"name": "open", "url": "http://localhost:9583/mcp"}],
+        )
+        load_mcp_tools_from_config(cfg)
+        _mock_mcp_stack["http"].assert_called_once_with(
+            url="http://localhost:9583/mcp",
+            token=None,
+        )
+
+    def test_stdio_server_does_not_get_token_kwarg(self, _mock_mcp_stack):
+        """StdioTransport doesn't take a token (unix-socket auth is OOB)."""
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        cfg = _make_mcp_cfg(
+            enabled=True,
+            servers=[
+                {"name": "local-mcp", "command": "mcp-server-foo", "args": ["--flag"]}
+            ],
+        )
+        load_mcp_tools_from_config(cfg)
+        _mock_mcp_stack["stdio"].assert_called_once_with(
+            command=["mcp-server-foo", "--flag"]
+        )
+
+
+class TestLoaderFiltering:
+    def test_allowed_names_filter_applied(self, _mock_mcp_stack):
+        """allowed_names limits the returned tools to that set."""
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        _mock_mcp_stack["provider"].return_value.discover.return_value = [
+            _fake_tool("alpha"),
+            _fake_tool("beta"),
+            _fake_tool("gamma"),
+        ]
+        cfg = _make_mcp_cfg(
+            enabled=True,
+            servers=[{"name": "x", "url": "http://x"}],
+        )
+        tools, _ = load_mcp_tools_from_config(cfg, allowed_names={"beta"})
+        assert [t.spec.name for t in tools] == ["beta"]
+
+    def test_include_tools_per_server(self, _mock_mcp_stack):
+        """Per-server include_tools restricts to just those names."""
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        _mock_mcp_stack["provider"].return_value.discover.return_value = [
+            _fake_tool("alpha"),
+            _fake_tool("beta"),
+        ]
+        cfg = _make_mcp_cfg(
+            enabled=True,
+            servers=[
+                {"name": "x", "url": "http://x", "include_tools": ["alpha"]}
+            ],
+        )
+        tools, _ = load_mcp_tools_from_config(cfg)
+        assert [t.spec.name for t in tools] == ["alpha"]
+
+    def test_exclude_tools_per_server(self, _mock_mcp_stack):
+        """Per-server exclude_tools drops the named tools."""
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        _mock_mcp_stack["provider"].return_value.discover.return_value = [
+            _fake_tool("alpha"),
+            _fake_tool("beta"),
+        ]
+        cfg = _make_mcp_cfg(
+            enabled=True,
+            servers=[
+                {"name": "x", "url": "http://x", "exclude_tools": ["alpha"]}
+            ],
+        )
+        tools, _ = load_mcp_tools_from_config(cfg)
+        assert [t.spec.name for t in tools] == ["beta"]
+
+
+class TestLoaderClientLifetime:
+    def test_returns_live_clients_for_caller_to_hold(self, _mock_mcp_stack):
+        """Critical lifetime contract — caller must hold `clients` so
+        the transports' httpx sessions stay open. (#461 adversarial
+        review caught this.) The list returned MUST contain a client
+        per successfully-initialized server."""
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        cfg = _make_mcp_cfg(
+            enabled=True,
+            servers=[
+                {"name": "s1", "url": "http://x1"},
+                {"name": "s2", "url": "http://x2"},
+            ],
+        )
+        _, clients = load_mcp_tools_from_config(cfg)
+        assert len(clients) == 2
+
+
+class TestLoaderFailureIsolation:
+    def test_one_server_failure_doesnt_abort_others(self, _mock_mcp_stack, caplog):
+        """When one server's initialize() raises, the loader logs and
+        moves on — the remaining servers still contribute tools."""
+        from openjarvis.mcp.loader import load_mcp_tools_from_config
+
+        # First initialize() raises, second succeeds
+        bad_client = MagicMock()
+        bad_client.initialize.side_effect = RuntimeError("can't reach server")
+        good_client = MagicMock()
+        good_client.initialize.return_value = None
+        _mock_mcp_stack["client"].side_effect = [bad_client, good_client]
+        _mock_mcp_stack["provider"].return_value.discover.return_value = [
+            _fake_tool("survivor")
+        ]
+
+        cfg = _make_mcp_cfg(
+            enabled=True,
+            servers=[
+                {"name": "broken", "url": "http://broken"},
+                {"name": "working", "url": "http://working"},
+            ],
+        )
+        with caplog.at_level("WARNING"):
+            tools, clients = load_mcp_tools_from_config(cfg)
+        assert [t.spec.name for t in tools] == ["survivor"]
+        assert len(clients) == 1
+        assert any("broken" in r.message for r in caplog.records)
diff --git a/tests/mcp/test_streamable_http_transport.py b/tests/mcp/test_streamable_http_transport.py
index 0bb9b979..e5d7b34c 100644
--- a/tests/mcp/test_streamable_http_transport.py
+++ b/tests/mcp/test_streamable_http_transport.py
@@ -100,6 +100,70 @@ class TestStreamableHTTPTransport:
         first_call_headers = mock_client.post.call_args[1]["headers"]
         assert "Mcp-Session-Id" not in first_call_headers
 
+    def test_authorization_header_with_token(self, _mock_httpx_client):
+        """Regression for #461 — token kwarg → Authorization: Bearer header."""
+        from openjarvis.mcp.transport import StreamableHTTPTransport
+
+        mock_client = _mock_httpx_client
+        mock_client.post.return_value = _make_http_response({})
+
+        transport = StreamableHTTPTransport(
+            "http://homeassistant.local:8123/mcp",
+            token="ha-long-lived-token-xyz",
+        )
+        transport.send(MCPRequest(method="tools/list", id=1))
+
+        headers = mock_client.post.call_args[1]["headers"]
+        assert headers["Authorization"] == "Bearer ha-long-lived-token-xyz"
+
+    def test_no_authorization_header_without_token(self, _mock_httpx_client):
+        """Backward compat — no token kwarg → no Authorization header."""
+        from openjarvis.mcp.transport import StreamableHTTPTransport
+
+        mock_client = _mock_httpx_client
+        mock_client.post.return_value = _make_http_response({})
+
+        transport = StreamableHTTPTransport("http://localhost:9583/mcp")
+        transport.send(MCPRequest(method="tools/list", id=1))
+
+        headers = mock_client.post.call_args[1]["headers"]
+        assert "Authorization" not in headers
+
+    def test_empty_token_does_not_send_header(self, _mock_httpx_client):
+        """token='' → no Authorization (empty/falsy tokens skip the header).
+
+        Matters because `cfg.get('token')` returns `''` if the user wrote
+        `token = ""` in config.toml. We don't want to send `Authorization:
+        Bearer ` (with a trailing space) — that's a malformed header that
+        most servers reject with a confusing 400 rather than 401.
+        """
+        from openjarvis.mcp.transport import StreamableHTTPTransport
+
+        mock_client = _mock_httpx_client
+        mock_client.post.return_value = _make_http_response({})
+
+        transport = StreamableHTTPTransport("http://localhost:9583/mcp", token="")
+        transport.send(MCPRequest(method="tools/list", id=1))
+
+        headers = mock_client.post.call_args[1]["headers"]
+        assert "Authorization" not in headers
+
+    def test_authorization_persists_across_requests(self, _mock_httpx_client):
+        """Authorization header must accompany every request, not just the first."""
+        from openjarvis.mcp.transport import StreamableHTTPTransport
+
+        mock_client = _mock_httpx_client
+        mock_client.post.return_value = _make_http_response({})
+
+        transport = StreamableHTTPTransport(
+            "http://localhost:9583/mcp", token="abc123"
+        )
+        for i in range(3):
+            transport.send(MCPRequest(method="tools/list", id=i))
+
+        for call in mock_client.post.call_args_list:
+            assert call[1]["headers"]["Authorization"] == "Bearer abc123"
+
     def test_connect_error_handling(self, _mock_httpx_client):
         """httpx.ConnectError should be wrapped in RuntimeError."""
         import httpx