diff --git a/README.md b/README.md
index 234fcf4f6..3d0c49750 100644
--- a/README.md
+++ b/README.md
@@ -672,6 +672,8 @@ uv run python -m deerflow.skills.review.cli ../skills/public/data-analysis --for
 
 Tools follow the same philosophy. DeerFlow comes with a core toolset — web search, web fetch, rendered web capture, file operations, bash execution — and supports custom tools via MCP servers and Python functions. Swap anything. Add anything.
 
+Advanced deployments can also extend the agent runtime itself by declaring zero-argument `AgentMiddleware` classes under `extensions.middlewares` in `config.yaml` or `extensions_config.json`. DeerFlow loads the same configured class list into the lead-agent and subagent pipelines after their built-in runtime middlewares and loop/token guards, but before the terminal-response/safety/clarification tail, so enterprise forks can add domain guardrails, tool-call governance, or observability hooks without patching the built-in middleware builders. Missing packages, invalid classes, and broken modules fail loudly at agent creation. Treat `config.yaml` and `extensions_config.json` as trusted operator-controlled files: middleware paths are code execution, just like custom tool, model, sandbox, guardrail, MCP server, and MCP interceptor declarations. Gateway skill/MCP toggle endpoints preserve this field but do not expose an API write path for `extensions.middlewares`. Per-context parameterization and separate lead-only/subagent-only middleware lists are not supported yet.
+
 Gateway-generated follow-up suggestions now normalize both plain-string model output and block/list-style rich content before parsing the JSON array response, so provider-specific content wrappers do not silently drop suggestions.
 
 The Web UI composer can polish draft input before sending. The rewrite runs as a short Gateway LLM request using the `input_polish` model configuration, keeps slash skill prefixes such as `/data-analysis`, and only replaces the local draft after the user clicks the polish button; it does not create a thread run or persist a message.
diff --git a/backend/AGENTS.md b/backend/AGENTS.md
index dcea77f76..398fc0d8b 100644
--- a/backend/AGENTS.md
+++ b/backend/AGENTS.md
@@ -262,10 +262,11 @@ Before changing a later authorization phase, read the [authorization RFC](../doc
 27. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce both the per-response concurrency limit (`max_concurrent_subagents`, clamped to 2-4) and the per-run total delegation cap (`max_total_subagents` runtime override or `subagents.max_total_per_run`, default 6, clamped to 1-50). The total cap counts current-run entries in the durable delegation ledger (entries are tagged with `run_id` when captured), so repeated planning checkpoints in one run cannot keep launching legal-sized batches indefinitely, while later user turns in the same thread get a fresh run budget. If the cap is exhausted, the middleware strips remaining `task` calls, forces `finish_reason="stop"`, and appends a visible limit note so the run can synthesize existing results instead of ending with an empty tool-call response.
 28. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer; stamps `loop_capped` via `consume_stop_reason` (#3875 Phase 2), symmetric to `TokenBudgetMiddleware`
 29. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits
-30. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the terminal-response/safety/clarification tail
-31. **TerminalResponseMiddleware** - When a provider returns an empty terminal `AIMessage` after tool execution, injects a hidden recovery prompt and retries the model once; a second empty response is replaced in checkpoint state by a visible error fallback marked for the run worker, so the run finishes as an error instead of a silent success
-32. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after terminal-response/custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first
-33. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for allowlisted clarification `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context.
+30. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before config-declared extensions and the terminal-response/safety/clarification tail
+31. **Configured extension middlewares** - *(optional, if `extensions.middlewares` is set in `config.yaml` or `extensions_config.json`)* Zero-argument `AgentMiddleware` classes loaded from `module.path:ClassName` entries via `deerflow.reflection.resolve_class`. Missing packages, invalid classes, and broken modules fail loudly at agent creation. These run after built-ins/programmatic custom middleware and after the lead/subagent loop/token guards, but before the terminal-response/safety/clarification tail; subagents receive the same configured extension middleware class list before their safety tail. Treat these files as trusted operator config because middleware paths instantiate arbitrary code. Gateway skill/MCP toggle endpoints preserve this field through `to_file_dict()` but must not add a write path for `extensions.middlewares` without an explicit trust-boundary review. Lead-only vs subagent-only middleware lists and per-context constructor parameters are not expressible in this MVP.
+32. **TerminalResponseMiddleware** - When a provider returns an empty terminal `AIMessage` after tool execution, injects a hidden recovery prompt and retries the model once; a second empty response is replaced in checkpoint state by a visible error fallback marked for the run worker, so the run finishes as an error instead of a silent success
+33. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after terminal-response/custom/configured middlewares so LangChain's reverse-order `after_model` dispatch runs it first
+34. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for allowlisted clarification `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context.
 
 ### Configuration System
 
@@ -787,8 +788,9 @@ Config is env-driven like the others — `MonocleTracingConfig`, built in `get_t
 - `mcpServers` - Map of server name → config (enabled, type, command, args, env, url, headers, oauth, description, `routing`, `tools`, `tool_call_timeout`). `routing.mode="prefer"` emits `<mcp_routing_hints>` prompt guidance; if `tool_search` defers the hinted tool, `McpRoutingMiddleware` can also auto-promote matching deferred schemas before the model call. It does not hard-disable other tools.
 - `tool_search.auto_promote_top_k` - Global MCP routing auto-promote breadth. Default `3`, clamped to `1..5`; applies only when `tool_search.enabled=true` and only to deferred MCP tools with `routing.mode="prefer"` and non-empty keywords. For lead agents the deferred catalog is built from the full configured MCP set; auto-promotion never grants authority because an active skill's runtime policy still filters model-visible schemas, `tool_search` results, and execution.
 - `skills` - Map of skill name → state (enabled)
+- `middlewares` - Zero-argument `AgentMiddleware` class paths for lead and subagent runtime extension. `config.yaml -> extensions` can override these fields after validation; overrides are replace-per-field, not list concatenation.
 
-Both can be modified at runtime via Gateway API endpoints or `DeerFlowClient` methods.
+Gateway API endpoints and `DeerFlowClient` methods can modify MCP servers and skill state at runtime; `middlewares` remains an operator-controlled config-file extension point.
 
 ### Embedded Client (`packages/harness/deerflow/client.py`)
 
diff --git a/backend/app/gateway/routers/skills.py b/backend/app/gateway/routers/skills.py
index caee64563..58eeebf0d 100644
--- a/backend/app/gateway/routers/skills.py
+++ b/backend/app/gateway/routers/skills.py
@@ -443,10 +443,7 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque
             extensions_config = get_extensions_config()
             extensions_config.skills[skill_name] = SkillStateConfig(enabled=body.enabled)
 
-            config_data = {
-                "mcpServers": {name: server.model_dump() for name, server in extensions_config.mcp_servers.items()},
-                "skills": {name: {"enabled": skill_config.enabled} for name, skill_config in extensions_config.skills.items()},
-            }
+            config_data = extensions_config.to_file_dict()
 
             with open(config_path, "w", encoding="utf-8") as f:
                 json.dump(config_data, f, indent=2)
@@ -466,10 +463,7 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque
                     config_path = Path.cwd().parent / "extensions_config.json"
                 extensions_config = get_extensions_config()
                 extensions_config.skills[skill_name] = SkillStateConfig(enabled=body.enabled)
-                config_data = {
-                    "mcpServers": {name: server.model_dump() for name, server in extensions_config.mcp_servers.items()},
-                    "skills": {name: {"enabled": skill_config.enabled} for name, skill_config in extensions_config.skills.items()},
-                }
+                config_data = extensions_config.to_file_dict()
                 with open(config_path, "w", encoding="utf-8") as f:
                     json.dump(config_data, f, indent=2)
                 reload_extensions_config()
diff --git a/backend/packages/harness/deerflow/agents/lead_agent/agent.py b/backend/packages/harness/deerflow/agents/lead_agent/agent.py
index fcf93b0e2..29d5848ca 100644
--- a/backend/packages/harness/deerflow/agents/lead_agent/agent.py
+++ b/backend/packages/harness/deerflow/agents/lead_agent/agent.py
@@ -33,6 +33,7 @@ from langchain_core.runnables import RunnableConfig
 
 from deerflow.agents.lead_agent.prompt import apply_prompt_template
 from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware
+from deerflow.agents.middlewares.configured_extensions import load_configured_extension_middlewares
 from deerflow.agents.middlewares.loop_detection_middleware import LoopDetectionMiddleware
 from deerflow.agents.middlewares.memory_middleware import MemoryMiddleware
 from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware
@@ -403,6 +404,10 @@ def build_middlewares(
     if custom_middlewares:
         middlewares.extend(custom_middlewares)
 
+    configured_middlewares = load_configured_extension_middlewares(resolved_app_config)
+    if configured_middlewares:
+        middlewares.extend(configured_middlewares)
+
     # A provider may return an empty AIMessage after tool execution. Retry the
     # final response once, then persist a visible error fallback rather than
     # allowing LangChain's no-tool-call router to end a silent successful run.
@@ -410,9 +415,9 @@ def build_middlewares(
 
     # SafetyFinishReasonMiddleware — suppress tool execution when the provider
     # safety-terminated the response. Registered after the terminal-response
-    # and custom middlewares so LangChain's reverse-order after_model dispatch
-    # runs Safety first; cleared tool_calls then flow through the remaining
-    # accounting/terminal guards without firing extra alarms.
+    # and custom/configured middlewares so LangChain's reverse-order after_model
+    # dispatch runs Safety first; cleared tool_calls then flow through the
+    # remaining accounting/terminal guards without firing extra alarms.
     safety_config = resolved_app_config.safety_finish_reason
     if safety_config.enabled:
         middlewares.append(SafetyFinishReasonMiddleware.from_config(safety_config))
diff --git a/backend/packages/harness/deerflow/agents/middlewares/configured_extensions.py b/backend/packages/harness/deerflow/agents/middlewares/configured_extensions.py
new file mode 100644
index 000000000..ef811426f
--- /dev/null
+++ b/backend/packages/harness/deerflow/agents/middlewares/configured_extensions.py
@@ -0,0 +1,34 @@
+"""Config-declared agent middleware loading."""
+
+import logging
+from typing import TYPE_CHECKING
+
+from langchain.agents.middleware import AgentMiddleware
+
+from deerflow.reflection import resolve_class
+
+if TYPE_CHECKING:
+    from deerflow.config.app_config import AppConfig
+
+logger = logging.getLogger(__name__)
+
+
+def load_configured_extension_middlewares(app_config: "AppConfig") -> list[AgentMiddleware]:
+    """Instantiate config-declared agent middlewares.
+
+    Each entry is a zero-argument ``AgentMiddleware`` class path in
+    ``module.path:ClassName`` format. Import, attribute, and subclass validation
+    intentionally go through the shared reflection resolver so failures carry
+    the same actionable dependency hints as models, tools, sandbox providers,
+    and guardrail providers.
+    """
+    middlewares: list[AgentMiddleware] = []
+    for middleware_path in list(app_config.extensions.middlewares or []):
+        middleware_cls = resolve_class(middleware_path, AgentMiddleware)
+        try:
+            middleware = middleware_cls()
+        except Exception:
+            logger.exception("Failed to instantiate configured extension middleware %s", middleware_path)
+            raise
+        middlewares.append(middleware)
+    return middlewares
diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py
index e50eae884..d82dcba4e 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py
@@ -369,6 +369,10 @@ def build_subagent_runtime_middlewares(
 
         middlewares.append(TokenBudgetMiddleware.from_config(token_budget_config))
 
+    from deerflow.agents.middlewares.configured_extensions import load_configured_extension_middlewares
+
+    middlewares.extend(load_configured_extension_middlewares(app_config))
+
     # Same provider safety-termination guard the lead agent uses — subagents
     # are equally exposed to truncated tool_calls returned with
     # finish_reason=content_filter (and friends), and the bad call would then
diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py
index 1046e9afa..0acfd05a9 100644
--- a/backend/packages/harness/deerflow/client.py
+++ b/backend/packages/harness/deerflow/client.py
@@ -1117,10 +1117,8 @@ class DeerFlowClient:
 
         current_config = get_extensions_config()
 
-        config_data = {
-            "mcpServers": mcp_servers,
-            "skills": {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()},
-        }
+        config_data = current_config.to_file_dict()
+        config_data["mcpServers"] = mcp_servers
 
         self._atomic_write_json(config_path, config_data)
 
@@ -1186,10 +1184,7 @@ class DeerFlowClient:
             extensions_config = get_extensions_config()
             extensions_config.skills[name] = SkillStateConfig(enabled=enabled)
 
-            config_data = {
-                "mcpServers": {n: s.model_dump() for n, s in extensions_config.mcp_servers.items()},
-                "skills": {n: {"enabled": sc.enabled} for n, sc in extensions_config.skills.items()},
-            }
+            config_data = extensions_config.to_file_dict()
 
             self._atomic_write_json(config_path, config_data)
             reload_extensions_config()
@@ -1206,10 +1201,7 @@ class DeerFlowClient:
                     raise FileNotFoundError("Cannot locate extensions_config.json. Set DEER_FLOW_EXTENSIONS_CONFIG_PATH or ensure it exists in the project root.")
                 extensions_config = get_extensions_config()
                 extensions_config.skills[name] = SkillStateConfig(enabled=enabled)
-                config_data = {
-                    "mcpServers": {n: s.model_dump() for n, s in extensions_config.mcp_servers.items()},
-                    "skills": {n: {"enabled": sc.enabled} for n, sc in extensions_config.skills.items()},
-                }
+                config_data = extensions_config.to_file_dict()
                 self._atomic_write_json(config_path, config_data)
                 reload_extensions_config()
 
diff --git a/backend/packages/harness/deerflow/config/app_config.py b/backend/packages/harness/deerflow/config/app_config.py
index ef1758fe9..66ee48b2e 100644
--- a/backend/packages/harness/deerflow/config/app_config.py
+++ b/backend/packages/harness/deerflow/config/app_config.py
@@ -378,9 +378,17 @@ class AppConfig(BaseModel):
         if "circuit_breaker" in config_data:
             config_data["circuit_breaker"] = config_data["circuit_breaker"]
 
-        # Load extensions config separately (it's in a different file)
+        # Load extensions config separately (it's in a different file), while
+        # preserving any config.yaml-backed extension fields. config.yaml wins
+        # when it explicitly declares a field because those values are part of
+        # the main AppConfig hot-reload contract.
+        yaml_extensions = config_data.get("extensions")
         extensions_config = ExtensionsConfig.from_file()
-        config_data["extensions"] = extensions_config.model_dump()
+        extensions_data = extensions_config.model_dump(by_alias=True)
+        if isinstance(yaml_extensions, Mapping):
+            yaml_extensions_config = ExtensionsConfig.model_validate(yaml_extensions)
+            extensions_data.update(yaml_extensions_config.model_dump(by_alias=True, exclude_unset=True))
+        config_data["extensions"] = extensions_data
 
         result = cls.model_validate(config_data)
         if not result.models:
diff --git a/backend/packages/harness/deerflow/config/extensions_config.py b/backend/packages/harness/deerflow/config/extensions_config.py
index 2043bf5be..1e886387f 100644
--- a/backend/packages/harness/deerflow/config/extensions_config.py
+++ b/backend/packages/harness/deerflow/config/extensions_config.py
@@ -132,6 +132,10 @@ class SkillStateConfig(BaseModel):
 class ExtensionsConfig(BaseModel):
     """Unified configuration for MCP servers and skills."""
 
+    middlewares: list[str] = Field(
+        default_factory=list,
+        description="AgentMiddleware class paths loaded into the lead-agent middleware chain. Each entry uses 'module.path:ClassName'.",
+    )
     mcp_servers: dict[str, McpServerConfig] = Field(
         default_factory=dict,
         description="Map of MCP server name to configuration",
@@ -143,6 +147,10 @@ class ExtensionsConfig(BaseModel):
     )
     model_config = ConfigDict(extra="allow", populate_by_name=True)
 
+    def to_file_dict(self) -> dict[str, Any]:
+        """Serialize in the public extensions_config.json shape."""
+        return self.model_dump(by_alias=True)
+
     @classmethod
     def resolve_config_path(cls, config_path: str | None = None) -> Path | None:
         """Resolve the extensions config file path.
diff --git a/backend/tests/test_app_config_reload.py b/backend/tests/test_app_config_reload.py
index a5a8dd7a1..febe3ac27 100644
--- a/backend/tests/test_app_config_reload.py
+++ b/backend/tests/test_app_config_reload.py
@@ -104,6 +104,14 @@ def _write_extensions_config(path: Path) -> None:
     path.write_text(json.dumps({"mcpServers": {}, "skills": {}}), encoding="utf-8")
 
 
+def test_config_example_does_not_enable_empty_extensions_block_by_default():
+    config_example_path = Path(__file__).resolve().parents[2] / "config.example.yaml"
+
+    config_data = yaml.safe_load(config_example_path.read_text(encoding="utf-8"))
+
+    assert "extensions" not in config_data
+
+
 def test_app_config_defaults_missing_database_to_sqlite(tmp_path, monkeypatch):
     config_path = tmp_path / "config.yaml"
     extensions_path = tmp_path / "extensions_config.json"
@@ -118,6 +126,79 @@ def test_app_config_defaults_missing_database_to_sqlite(tmp_path, monkeypatch):
     assert config.database.sqlite_dir == ".deer-flow/data"
 
 
+def test_app_config_preserves_config_yaml_extension_middlewares(tmp_path, monkeypatch):
+    config_path = tmp_path / "config.yaml"
+    extensions_path = tmp_path / "extensions_config.json"
+    extensions_path.write_text(
+        json.dumps({"mcpServers": {}, "skills": {}, "middlewares": ["pkg.from_file:FileMiddleware"]}),
+        encoding="utf-8",
+    )
+    _write_config_with_sections(
+        config_path,
+        {
+            "extensions": {
+                "middlewares": ["pkg.from_yaml:YamlMiddleware"],
+            }
+        },
+    )
+    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
+
+    config = AppConfig.from_file(str(config_path))
+
+    assert config.extensions.middlewares == ["pkg.from_yaml:YamlMiddleware"]
+
+
+def test_app_config_normalizes_config_yaml_extension_aliases_before_override(tmp_path, monkeypatch):
+    config_path = tmp_path / "config.yaml"
+    extensions_path = tmp_path / "extensions_config.json"
+    extensions_path.write_text(
+        json.dumps(
+            {
+                "mcpServers": {
+                    "from-file": {
+                        "command": "file-mcp",
+                    }
+                },
+                "skills": {},
+            }
+        ),
+        encoding="utf-8",
+    )
+    _write_config_with_sections(
+        config_path,
+        {
+            "extensions": {
+                "mcp_servers": {
+                    "from-yaml": {
+                        "command": "yaml-mcp",
+                    }
+                },
+            }
+        },
+    )
+    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
+
+    config = AppConfig.from_file(str(config_path))
+
+    assert set(config.extensions.mcp_servers) == {"from-yaml"}
+    assert config.extensions.mcp_servers["from-yaml"].command == "yaml-mcp"
+
+
+def test_app_config_loads_extension_middlewares_from_extensions_config(tmp_path, monkeypatch):
+    config_path = tmp_path / "config.yaml"
+    extensions_path = tmp_path / "extensions_config.json"
+    extensions_path.write_text(
+        json.dumps({"mcpServers": {}, "skills": {}, "middlewares": ["pkg.from_file:FileMiddleware"]}),
+        encoding="utf-8",
+    )
+    _write_config_with_sections(config_path)
+    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path))
+
+    config = AppConfig.from_file(str(config_path))
+
+    assert config.extensions.middlewares == ["pkg.from_file:FileMiddleware"]
+
+
 def test_app_config_defaults_empty_database_to_sqlite(tmp_path, monkeypatch):
     config_path = tmp_path / "config.yaml"
     extensions_path = tmp_path / "extensions_config.json"
diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py
index 69238d6e3..ed08049eb 100644
--- a/backend/tests/test_client.py
+++ b/backend/tests/test_client.py
@@ -19,6 +19,7 @@ from app.gateway.routers.skills import SkillInstallResponse, SkillResponse, Skil
 from app.gateway.routers.threads import ThreadGoalResponse
 from app.gateway.routers.uploads import UploadResponse
 from deerflow.client import DeerFlowClient
+from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig
 from deerflow.config.paths import Paths
 from deerflow.skills.types import SkillCategory
 from deerflow.uploads.manager import PathTraversalError
@@ -1429,13 +1430,9 @@ class TestMcpConfig:
 
     def test_update_mcp_config(self, client):
         # Set up current config with skills
-        current_config = MagicMock()
-        current_config.skills = {}
+        current_config = ExtensionsConfig()
 
-        reloaded_server = MagicMock()
-        reloaded_server.model_dump.return_value = {"enabled": True, "type": "sse"}
-        reloaded_config = MagicMock()
-        reloaded_config.mcp_servers = {"new-server": reloaded_server}
+        reloaded_config = ExtensionsConfig(mcp_servers={"new-server": McpServerConfig(enabled=True, type="sse")})
 
         with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
             json.dump({}, f)
@@ -1495,9 +1492,7 @@ class TestSkillsManagement:
         skill = self._make_skill(enabled=True)
         updated_skill = self._make_skill(enabled=False)
 
-        ext_config = MagicMock()
-        ext_config.mcp_servers = {}
-        ext_config.skills = {}
+        ext_config = ExtensionsConfig()
 
         with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
             json.dump({}, f)
@@ -2219,13 +2214,9 @@ class TestScenarioConfigManagement:
             config_file.write_text("{}")
 
             # --- MCP update ---
-            current_config = MagicMock()
-            current_config.skills = {}
+            current_config = ExtensionsConfig()
 
-            reloaded_server = MagicMock()
-            reloaded_server.model_dump.return_value = {"enabled": True, "type": "sse"}
-            reloaded_config = MagicMock()
-            reloaded_config.mcp_servers = {"my-mcp": reloaded_server}
+            reloaded_config = ExtensionsConfig(mcp_servers={"my-mcp": McpServerConfig(enabled=True, type="sse")})
 
             client._agent = MagicMock()  # Simulate existing agent
             with (
@@ -2252,9 +2243,7 @@ class TestScenarioConfigManagement:
             toggled.category = "custom"
             toggled.enabled = False
 
-            ext_config = MagicMock()
-            ext_config.mcp_servers = {}
-            ext_config.skills = {}
+            ext_config = ExtensionsConfig()
 
             client._agent = MagicMock()  # Simulate re-created agent
             with (
@@ -2762,20 +2751,17 @@ class TestGatewayConformance:
         assert "test" in parsed.mcp_servers
 
     def test_update_mcp_config(self, client, tmp_path):
-        server = MagicMock()
-        server.model_dump.return_value = {
-            "enabled": True,
-            "type": "stdio",
-            "command": "npx",
-            "args": [],
-            "env": {},
-            "url": None,
-            "headers": {},
-            "description": "",
-        }
-        ext_config = MagicMock()
-        ext_config.mcp_servers = {"srv": server}
-        ext_config.skills = {}
+        server = McpServerConfig(
+            enabled=True,
+            type="stdio",
+            command="npx",
+            args=[],
+            env={},
+            url=None,
+            headers={},
+            description="",
+        )
+        ext_config = ExtensionsConfig(mcp_servers={"srv": server})
 
         config_file = tmp_path / "extensions_config.json"
         config_file.write_text("{}")
@@ -2785,7 +2771,7 @@ class TestGatewayConformance:
             patch("deerflow.client.ExtensionsConfig.resolve_config_path", return_value=config_file),
             patch("deerflow.client.reload_extensions_config", return_value=ext_config),
         ):
-            result = client.update_mcp_config({"srv": server.model_dump.return_value})
+            result = client.update_mcp_config({"srv": server.model_dump()})
 
         parsed = McpConfigResponse(**result)
         assert "srv" in parsed.mcp_servers
@@ -3544,10 +3530,8 @@ class TestBugAgentInvalidationInconsistency:
         client._agent = MagicMock()
         client._agent_config_key = ("model", True, False, False)
 
-        current_config = MagicMock()
-        current_config.skills = {}
-        reloaded = MagicMock()
-        reloaded.mcp_servers = {}
+        current_config = ExtensionsConfig()
+        reloaded = ExtensionsConfig()
 
         with tempfile.TemporaryDirectory() as tmp:
             config_file = Path(tmp) / "ext.json"
diff --git a/backend/tests/test_client_e2e.py b/backend/tests/test_client_e2e.py
index 5e2e02186..3ece42b03 100644
--- a/backend/tests/test_client_e2e.py
+++ b/backend/tests/test_client_e2e.py
@@ -702,7 +702,7 @@ class TestConfigManagement:
         """update_mcp_config() writes extensions_config.json and invalidates the agent."""
         # Set up a writable extensions_config.json
         config_file = tmp_path / "extensions_config.json"
-        config_file.write_text(json.dumps({"mcpServers": {}, "skills": {}}))
+        config_file.write_text(json.dumps({"mcpServers": {}, "skills": {}, "middlewares": ["pkg:Middleware"]}))
         monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_file))
 
         # Force reload so the singleton picks up our test file
@@ -725,11 +725,12 @@ class TestConfigManagement:
         # File should be written
         written = json.loads(config_file.read_text())
         assert "test-server" in written["mcpServers"]
+        assert written["middlewares"] == ["pkg:Middleware"]
 
     def test_update_skill_writes_and_invalidates(self, e2e_env, tmp_path, monkeypatch):
         """update_skill() writes extensions_config.json and invalidates the agent."""
         config_file = tmp_path / "extensions_config.json"
-        config_file.write_text(json.dumps({"mcpServers": {}, "skills": {}}))
+        config_file.write_text(json.dumps({"mcpServers": {}, "skills": {}, "middlewares": ["pkg:Middleware"]}))
         monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_file))
 
         from deerflow.config.extensions_config import reload_extensions_config
@@ -749,6 +750,8 @@ class TestConfigManagement:
         result = c.update_skill(skill_name, enabled=False)
         assert result["name"] == skill_name
         assert result["enabled"] is False
+        written = json.loads(config_file.read_text())
+        assert written["middlewares"] == ["pkg:Middleware"]
 
         # Agent should be invalidated
         assert c._agent is None
diff --git a/backend/tests/test_lead_agent_model_resolution.py b/backend/tests/test_lead_agent_model_resolution.py
index bc029f2cf..e9a99a9a5 100644
--- a/backend/tests/test_lead_agent_model_resolution.py
+++ b/backend/tests/test_lead_agent_model_resolution.py
@@ -9,6 +9,7 @@ from unittest.mock import MagicMock
 
 import pytest
 from langchain.agents import create_agent
+from langchain.agents.middleware import AgentMiddleware
 from langchain_core.language_models import BaseChatModel
 from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
 from langchain_core.outputs import ChatGeneration, ChatResult
@@ -20,6 +21,7 @@ from deerflow.agents.middlewares.loop_detection_middleware import LoopDetectionM
 from deerflow.agents.middlewares.subagent_limit_middleware import SubagentLimitMiddleware
 from deerflow.agents.thread_state import ThreadState
 from deerflow.config.app_config import AppConfig
+from deerflow.config.extensions_config import ExtensionsConfig
 from deerflow.config.loop_detection_config import LoopDetectionConfig
 from deerflow.config.memory_config import MemoryConfig
 from deerflow.config.model_config import ModelConfig
@@ -102,6 +104,23 @@ def _make_model(name: str, *, supports_thinking: bool) -> ModelConfig:
     )
 
 
+class ConfiguredGuardMiddleware(AgentMiddleware):
+    pass
+
+
+class ConfiguredAuditMiddleware(AgentMiddleware):
+    pass
+
+
+class ConfiguredInitFailureMiddleware(AgentMiddleware):
+    def __init__(self) -> None:
+        raise RuntimeError("configured middleware init failed")
+
+
+class ConfiguredNonMiddleware:
+    pass
+
+
 def test_make_lead_agent_signature_matches_langgraph_server_factory_abi():
     assert list(inspect.signature(lead_agent_module.make_lead_agent).parameters) == ["config"]
 
@@ -671,6 +690,42 @@ def test_build_middlewares_omits_loop_detection_when_disabled(monkeypatch):
     assert not any(isinstance(m, LoopDetectionMiddleware) for m in middlewares)
 
 
+def test_build_middlewares_injects_configured_extension_middlewares(monkeypatch):
+    app_config = _make_app_config(
+        [_make_model("safe-model", supports_thinking=False)],
+        loop_detection=LoopDetectionConfig(enabled=False),
+    )
+    app_config.extensions = ExtensionsConfig(
+        middlewares=[
+            f"{__name__}:ConfiguredGuardMiddleware",
+            f"{__name__}:ConfiguredAuditMiddleware",
+        ]
+    )
+    manual_middleware = MagicMock()
+
+    monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
+    monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
+    monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None)
+    monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
+
+    middlewares = lead_agent_module.build_middlewares(
+        {"configurable": {"is_plan_mode": False, "subagent_enabled": False}},
+        model_name="safe-model",
+        custom_middlewares=[manual_middleware],
+        app_config=app_config,
+    )
+
+    middleware_types = [type(m).__name__ for m in middlewares]
+    assert middleware_types[-5:] == [
+        "ConfiguredGuardMiddleware",
+        "ConfiguredAuditMiddleware",
+        "TerminalResponseMiddleware",
+        "SafetyFinishReasonMiddleware",
+        "ClarificationMiddleware",
+    ]
+    assert middlewares[middleware_types.index("ConfiguredGuardMiddleware") - 1] is manual_middleware
+
+
 def test_build_middlewares_passes_subagent_total_limit_from_app_config(monkeypatch):
     app_config = _make_app_config(
         [_make_model("safe-model", supports_thinking=False)],
@@ -723,6 +778,86 @@ def test_build_middlewares_allows_runtime_subagent_total_limit_override(monkeypa
     assert limit.max_total == 5
 
 
+def test_build_middlewares_rejects_invalid_configured_extension_middleware(monkeypatch):
+    app_config = _make_app_config(
+        [_make_model("safe-model", supports_thinking=False)],
+        loop_detection=LoopDetectionConfig(enabled=False),
+    )
+    app_config.extensions = ExtensionsConfig(middlewares=[f"{__name__}:_make_model"])
+
+    monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
+    monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
+    monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None)
+    monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
+
+    with pytest.raises(ValueError, match="not an instance of type"):
+        lead_agent_module.build_middlewares(
+            {"configurable": {"is_plan_mode": False, "subagent_enabled": False}},
+            model_name="safe-model",
+            app_config=app_config,
+        )
+
+
+def test_build_middlewares_rejects_configured_extension_class_with_wrong_base(monkeypatch):
+    app_config = _make_app_config(
+        [_make_model("safe-model", supports_thinking=False)],
+        loop_detection=LoopDetectionConfig(enabled=False),
+    )
+    app_config.extensions = ExtensionsConfig(middlewares=[f"{__name__}:ConfiguredNonMiddleware"])
+
+    monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
+    monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
+    monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None)
+    monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
+
+    with pytest.raises(ValueError, match="is not a subclass of AgentMiddleware"):
+        lead_agent_module.build_middlewares(
+            {"configurable": {"is_plan_mode": False, "subagent_enabled": False}},
+            model_name="safe-model",
+            app_config=app_config,
+        )
+
+
+def test_build_middlewares_reraises_configured_extension_instantiation_failure(monkeypatch):
+    app_config = _make_app_config(
+        [_make_model("safe-model", supports_thinking=False)],
+        loop_detection=LoopDetectionConfig(enabled=False),
+    )
+    app_config.extensions = ExtensionsConfig(middlewares=[f"{__name__}:ConfiguredInitFailureMiddleware"])
+
+    monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
+    monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
+    monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None)
+    monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
+
+    with pytest.raises(RuntimeError, match="configured middleware init failed"):
+        lead_agent_module.build_middlewares(
+            {"configurable": {"is_plan_mode": False, "subagent_enabled": False}},
+            model_name="safe-model",
+            app_config=app_config,
+        )
+
+
+def test_build_middlewares_rejects_missing_configured_extension_module(monkeypatch):
+    app_config = _make_app_config(
+        [_make_model("safe-model", supports_thinking=False)],
+        loop_detection=LoopDetectionConfig(enabled=False),
+    )
+    app_config.extensions = ExtensionsConfig(middlewares=["definitely_missing_pkg.middlewares_typo:GuardMiddleware"])
+
+    monkeypatch.setattr(lead_agent_module, "get_app_config", lambda: app_config)
+    monkeypatch.setattr(lead_agent_module, "build_lead_runtime_middlewares", lambda *, app_config, lazy_init=True: [])
+    monkeypatch.setattr(lead_agent_module, "_create_summarization_middleware", lambda *, app_config=None: None)
+    monkeypatch.setattr(lead_agent_module, "_create_todo_list_middleware", lambda is_plan_mode: None)
+
+    with pytest.raises(ImportError, match="Could not import module definitely_missing_pkg.middlewares_typo"):
+        lead_agent_module.build_middlewares(
+            {"configurable": {"is_plan_mode": False, "subagent_enabled": False}},
+            model_name="safe-model",
+            app_config=app_config,
+        )
+
+
 def test_create_summarization_middleware_uses_configured_model_alias(monkeypatch):
     app_config = _make_app_config([_make_model("model-masswork", supports_thinking=False)])
     app_config.summarization = SummarizationConfig(enabled=True, model_name="model-masswork")
diff --git a/backend/tests/test_skills_router_authz.py b/backend/tests/test_skills_router_authz.py
index 4f1eb05d2..955705e81 100644
--- a/backend/tests/test_skills_router_authz.py
+++ b/backend/tests/test_skills_router_authz.py
@@ -19,6 +19,7 @@ These tests pin the access-control boundary: a normal authenticated
 
 from __future__ import annotations
 
+import json
 from types import SimpleNamespace
 from uuid import uuid4
 
@@ -141,7 +142,18 @@ def test_enable_toggle_allowed_for_admin(monkeypatch, tmp_path):
 
     app = _make_app(system_role="admin")
     monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: SimpleNamespace(load_skills=_load_skills))
-    monkeypatch.setattr(skills_router, "get_extensions_config", lambda: SimpleNamespace(mcp_servers={}, skills={}))
+    from deerflow.config.extensions_config import ExtensionsConfig
+
+    monkeypatch.setattr(
+        skills_router,
+        "get_extensions_config",
+        lambda: ExtensionsConfig(
+            mcp_servers={},
+            skills={},
+            middlewares=["pkg:Middleware"],
+            mcpInterceptors=["pkg.interceptor:build"],
+        ),
+    )
     monkeypatch.setattr(skills_router, "reload_extensions_config", lambda: None)
     monkeypatch.setattr(skills_router.ExtensionsConfig, "resolve_config_path", staticmethod(lambda: config_path))
 
@@ -152,3 +164,6 @@ def test_enable_toggle_allowed_for_admin(monkeypatch, tmp_path):
     with TestClient(app) as client:
         resp = client.put("/api/skills/demo", json={"enabled": False})
         assert resp.status_code == 200, f"admin toggle should succeed, got {resp.status_code}"
+    written = json.loads(config_path.read_text(encoding="utf-8"))
+    assert written["middlewares"] == ["pkg:Middleware"]
+    assert written["mcpInterceptors"] == ["pkg.interceptor:build"]
diff --git a/backend/tests/test_tool_error_handling_middleware.py b/backend/tests/test_tool_error_handling_middleware.py
index 5c358c48f..515ac97aa 100644
--- a/backend/tests/test_tool_error_handling_middleware.py
+++ b/backend/tests/test_tool_error_handling_middleware.py
@@ -3,6 +3,7 @@ import sys
 from types import ModuleType, SimpleNamespace
 
 import pytest
+from langchain.agents.middleware import AgentMiddleware
 from langchain_core.messages import ToolMessage
 from langgraph.errors import GraphInterrupt
 
@@ -15,12 +16,17 @@ from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY
 from deerflow.agents.middlewares.view_image_middleware import ViewImageMiddleware
 from deerflow.config import summarization_config
 from deerflow.config.app_config import AppConfig, CircuitBreakerConfig
+from deerflow.config.extensions_config import ExtensionsConfig
 from deerflow.config.guardrails_config import GuardrailsConfig
 from deerflow.config.model_config import ModelConfig
 from deerflow.config.sandbox_config import SandboxConfig
 from deerflow.subagents.status_contract import SUBAGENT_ERROR_KEY, SUBAGENT_STATUS_KEY
 
 
+class ConfiguredSubagentMiddleware(AgentMiddleware):
+    pass
+
+
 def _request(name: str = "web_search", tool_call_id: str | None = "tc-1"):
     tool_call = {"name": name}
     if tool_call_id is not None:
@@ -560,6 +566,24 @@ def test_subagent_runtime_middlewares_attach_deferred_filter_when_setup_has_name
     assert filter_idx < safety_idx
 
 
+def test_subagent_runtime_middlewares_inject_configured_extension_middlewares(monkeypatch):
+    from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware
+
+    app_config = _make_app_config()
+    app_config.extensions = ExtensionsConfig(middlewares=[f"{__name__}:ConfiguredSubagentMiddleware"])
+    _stub_runtime_middleware_imports(monkeypatch)
+
+    middlewares = build_subagent_runtime_middlewares(app_config=app_config)
+
+    extension_idx = next(i for i, m in enumerate(middlewares) if isinstance(m, ConfiguredSubagentMiddleware))
+    safety_idx = next(i for i, m in enumerate(middlewares) if isinstance(m, SafetyFinishReasonMiddleware))
+    for guard_name in ("LoopDetectionMiddleware", "TokenBudgetMiddleware"):
+        guard_idx = next((i for i, m in enumerate(middlewares) if type(m).__name__ == guard_name), None)
+        if guard_idx is not None:
+            assert guard_idx < extension_idx
+    assert extension_idx < safety_idx
+
+
 def test_subagent_runtime_middlewares_place_mcp_routing_before_deferred_filter(monkeypatch):
     from deerflow.agents.middlewares.deferred_tool_filter_middleware import DeferredToolFilterMiddleware
     from deerflow.agents.middlewares.mcp_routing_middleware import McpRoutingMiddleware
diff --git a/config.example.yaml b/config.example.yaml
index df3d8c2ad..a61678b6e 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -30,6 +30,23 @@ logging:
     enabled: false
     format: text
 
+# ============================================================================
+# Agent Extensions
+# ============================================================================
+# Optional AgentMiddleware classes loaded into the lead and subagent runtime
+# middleware chains after built-in runtime middlewares, but before the
+# safety/clarification tail. Missing packages, invalid classes, and broken
+# modules fail loudly at agent creation with an actionable import error.
+# The same zero-argument class list applies to both lead and subagent runtimes;
+# lead-only vs subagent-only configuration is not expressible yet. Treat these
+# files as trusted operator config because middleware classes execute code.
+# Uncomment this block to define middlewares in config.yaml. Leaving it commented
+# lets extensions_config.json remain the source of truth for extension packages.
+# extensions:
+#   middlewares:
+#     - my_company.deerflow_middlewares:DomainGuardMiddleware
+#     - my_company.deerflow_middlewares:LatencyStampingMiddleware
+
 # ============================================================================
 # Tracing / Observability (Monocle)
 # ============================================================================
diff --git a/extensions_config.example.json b/extensions_config.example.json
index bb86f1cde..53f4e92d2 100644
--- a/extensions_config.example.json
+++ b/extensions_config.example.json
@@ -1,4 +1,5 @@
 {
+  "middlewares": [],
   "mcpInterceptors": [
     "my_package.mcp.auth:build_auth_interceptor"
   ],