Jiaqi Zhang (JiaqiZhang-Dev) · Nov 5, 2026 · Nov 5, 2026 · Nov 6, 2026 · Nov 6, 2026 · Nov 6, 2026
@@ -111,14 +111,16 @@ This reduces exposure to external contributors' free text. It is a heuristic, no
 
 ### C4 — External MCP spotlighting
 
-The [tool-output middleware](https://github.com/Azure/azure-sdk-tools/blob/main/tools/sdk-ai-bots/azure-sdk-qa-bot-agent/utils/tool_security.py) wraps nonempty external MCP string results in a labelled, JSON-escaped block:
+The [tool-output middleware](https://github.com/Azure/azure-sdk-tools/blob/main/tools/sdk-ai-bots/azure-sdk-qa-bot-agent/middleware/tool_output_security.py) wraps nonempty external tool results in a labelled, JSON-escaped block:
 
 ```text
 <untrusted_tool_output>
 "...json-escaped tool text..."
 </untrusted_tool_output>
 ```
 
+Text blocks in mixed-content results and session-memory read/search results receive the same wrapper. Bookkeeping operations remain unchanged. Large external results are stored in session-scoped file memory; the harness does not expose shared file access or local shell execution.
+
 This [spotlighting technique](https://arxiv.org/abs/2403.14720) cues the model to treat the content as reference data, not instructions. It does not guarantee resistance to injection. Native registered tools bypass the wrapper to preserve server-side decoding; their transcript, KB, trace, and web text remains untrusted.
 
 ### C5 — Platform content safety

@@ -13,6 +13,7 @@ COPY requirements.txt .
 RUN pip install --no-cache-dir --pre -r requirements.txt
 
 COPY config/ ./config/
+COPY middleware/ ./middleware/
 COPY models/ ./models/
 COPY prompts/ ./prompts/
 COPY services/ ./services/

@@ -5,6 +5,7 @@
 import os
 import sys
 from pathlib import Path
+from typing import cast
 
 import yaml
 from dotenv import load_dotenv
@@ -17,11 +18,14 @@
 os.environ.setdefault("ENABLE_SENSITIVE_DATA", "true")
 
 from agent_framework import Agent
+from agent_framework import AggregatingSkillsSource
 from agent_framework import CompactionProvider
 from agent_framework import FileSkillsSource
+from agent_framework import InMemorySkillsSource
 from agent_framework import SkillsProvider
 from agent_framework import ToolResultCompactionStrategy
 from agent_framework_foundry_hosting import ResponsesHostServer
+from agent_framework.openai import OpenAIChatOptions
 
 import config.app_config as app_config
 from config.app_config import get as cfg
@@ -34,7 +38,7 @@
 from utils.azure_ai_foundry import get_agent_client, get_project_client
 from utils.azure_memory_store import ensure_user_memory_store
 from utils.memory_context_provider import MemoryContextProvider
-from utils.tool_security import ToolOutputSecurityMiddleware
+from middleware.tool_output_security import ToolOutputSecurityMiddleware
 
 logger = logging.getLogger(__name__)
 
@@ -107,7 +111,14 @@ async def _init_mcp(factory, **kwargs):
     if not skills:
         raise RuntimeError(f"No tenant skills configured for agent {agent_name!r}")
     skills_dir = Path(tenant_skills.__file__).parent / "azure_mcp_server"
-    skills.extend(await FileSkillsSource(skills_dir).get_skills())
+    skills_provider = SkillsProvider(
+        AggregatingSkillsSource(
+            [
+                InMemorySkillsSource(skills),
+                FileSkillsSource(skills_dir),
+            ]
+        )
+    )
 
     memory_provider = MemoryContextProvider(project_client)
     compaction_provider = CompactionProvider(
@@ -122,16 +133,23 @@ async def _init_mcp(factory, **kwargs):
         instructions=instructions,
         tools=tools,
         context_providers=[
-            SkillsProvider(skills),
+            skills_provider,
             memory_provider,
             compaction_provider,
         ],
-        middleware=[ToolOutputSecurityMiddleware()],
-        default_options={
-            "reasoning": {"effort": cfg("AI_FOUNDRY_AGENT_REASONING_EFFORT")},
-            "max_tool_calls": MAX_TOOL_CALLS_PER_TURN,
-            "include": ["web_search_call.action.sources"],
-        },
+        middleware=[
+            ToolOutputSecurityMiddleware(),
+        ],
+        default_options=cast(
+            OpenAIChatOptions,
+            {
+                "reasoning": {
+                    "effort": cfg("AI_FOUNDRY_AGENT_REASONING_EFFORT")
+                },
+                "max_tool_calls": MAX_TOOL_CALLS_PER_TURN,
+                "include": ["web_search_call.action.sources"],
+            },
+        ),
     )
 
     server = ResponsesHostServer(agent)

@@ -31,6 +31,7 @@ RUN pip install --no-cache-dir --pre -r requirements.txt
 
 # Copy shared modules
 COPY config/ ./config/
+COPY middleware/ ./middleware/
 COPY models/ ./models/
 COPY prompts/ ./prompts/
 COPY services/ ./services/

@@ -23,11 +23,12 @@
 load_dotenv(override=False)
 os.environ.setdefault("ENABLE_SENSITIVE_DATA", "true")
 
-from agent_framework import Agent
-from agent_framework import CompactionProvider
-from agent_framework import FileSkillsSource
-from agent_framework import SkillsProvider
-from agent_framework import ToolResultCompactionStrategy
+from agent_framework import (
+    AggregatingSkillsSource,
+    FileSkillsSource,
+    InMemorySkillsSource,
+    SkillsProvider,
+)
 from agent_framework_foundry_hosting import ResponsesHostServer
 
 import config.app_config as app_config
@@ -41,14 +42,15 @@
 from tools.azsdk_mcp_tools import create_azsdk_mcp_tool
 from tools.github_mcp_tools import create_github_mcp_tool
 from utils.azure_ai_foundry import (
+    create_hosted_chat_agent,
     get_agent_client,
     get_project_client,
 )
 from utils.azure_memory_store import (
     ensure_user_memory_store,
 )
 from utils.memory_context_provider import MemoryContextProvider
-from utils.tool_security import ToolOutputSecurityMiddleware
+from middleware.tool_output_security import ToolOutputSecurityMiddleware
 
 logger = logging.getLogger(__name__)
 
@@ -86,6 +88,11 @@ async def main() -> None:
         MAX_TOOL_CALL_ITERATIONS
     )
     agent_dir = Path(__file__).parent
+    harness_instructions = _load_instructions(
+        Path(app_config.__file__).resolve().parent.parent
+        / "prompts"
+        / "harness_instruction.md"
+    )
     instructions = _load_instructions(agent_dir / "instruction.md")
     with open(agent_dir / "agent.yaml", encoding="utf-8") as f:
         agent_config = yaml.safe_load(f)
@@ -100,16 +107,12 @@ async def main() -> None:
     knowledge_tools = KnowledgeTools()
     identity_tools = IdentityTools()
     web_tools = WebTools()
-    web_search_tool = agent_client.get_web_search_tool(
-        search_context_size="medium",
-    )
 
     tools = [
         knowledge_tools.search_knowledge_base,
         knowledge_tools.wiki_search,
         identity_tools.resolve_github_users,
         web_tools.web_fetch,
-        web_search_tool,
     ]
 
     # Parallelise slow async startup tasks to reduce cold-start latency.
@@ -147,35 +150,45 @@ async def _init_mcp(factory):
     # Memory context provider (memory store initializes in background; may not be ready yet)
     memory_provider = MemoryContextProvider(project_client)
 
-    # Compaction provider — compact history before and after each turn
-    compaction_provider = CompactionProvider(
-        before_strategy=ToolResultCompactionStrategy(keep_last_tool_call_groups=2),
-        after_strategy=ToolResultCompactionStrategy(keep_last_tool_call_groups=1),
-    )
-
     # Init Skills
     skills = create_tenant_skills(agent_name)
     skills_dir = _get_chat_skill_dir()
-    skills.extend(await FileSkillsSource(skills_dir).get_skills())
-    skills_provider = SkillsProvider(skills)
+    skills_provider = SkillsProvider(
+        AggregatingSkillsSource(
+            [
+                InMemorySkillsSource(skills),
+                FileSkillsSource(skill_paths=skills_dir),
+            ]
+        ),
+        disable_load_skill_approval=True,  # pyright: ignore[reportCallIssue]
+        disable_read_skill_resource_approval=True,  # pyright: ignore[reportCallIssue]
+        disable_run_skill_script_approval=True,  # pyright: ignore[reportCallIssue]
+    )
 
     reasoning_effort = cfg("AI_FOUNDRY_AGENT_REASONING_EFFORT")
-    agent = Agent(
+    agent = create_hosted_chat_agent(
         agent_client,
         name=agent_name,
-        id=agent_id,
+        agent_id=agent_id,
+        harness_instructions=harness_instructions,
         instructions=instructions,
         tools=tools,
-        context_providers=[skills_provider, memory_provider, compaction_provider],
-        middleware=[ToolOutputSecurityMiddleware()],
+        skills_provider=skills_provider,
+        memory_provider=memory_provider,
+        middleware=[
+            ToolOutputSecurityMiddleware(),
+        ],
         default_options={
             "reasoning": {"effort": reasoning_effort},
             "max_tool_calls": MAX_TOOL_CALLS_PER_TURN,
             "include": ["web_search_call.action.sources"],
         },
     )
 
-    server = ResponsesHostServer(agent)
+    server = ResponsesHostServer(
+        agent,
+        history_source="agent",
+    )
     await server.run_async()
 
 

@@ -31,6 +31,7 @@ RUN pip install --no-cache-dir --pre -r requirements.txt
 
 # Copy shared modules
 COPY config/ ./config/
+COPY middleware/ ./middleware/
 COPY models/ ./models/
 COPY prompts/ ./prompts/
 COPY services/ ./services/

@@ -50,7 +50,7 @@
     get_project_client,
 )
 from utils.azure_storage import create_blob_service_client
-from utils.tool_security import ToolOutputSecurityMiddleware
+from middleware.tool_output_security import ToolOutputSecurityMiddleware
 
 logger = logging.getLogger(__name__)
 
@@ -165,7 +165,9 @@ async def main() -> None:
         instructions=instructions,
         tools=tools,
         context_providers=[compaction_provider],
-        middleware=[ToolOutputSecurityMiddleware()],
+        middleware=[
+            ToolOutputSecurityMiddleware(),
+        ],
         default_options={
             "reasoning": {"effort": reasoning_effort},
             "max_tool_calls": MAX_TOOL_CALLS_PER_TURN,

@@ -0,0 +1 @@
+"""Agent middleware."""
@@ -0,0 +1,68 @@
+"""Tool-output security middleware (prompt-injection defence).
+
+Spotlights untrusted tool output as *data, not instructions* by wrapping it in
+a labelled, JSON-escaped delimiter. Native ``@tool`` functions and trusted
+harness tools are left unchanged.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Awaitable, Callable
+from copy import copy
+
+from agent_framework import Content, FunctionInvocationContext, FunctionMiddleware
+
+from tools import TOOL_REGISTRY
+
+_SPOTLIGHT_OPEN = "<untrusted_tool_output>"
+_SPOTLIGHT_CLOSE = "</untrusted_tool_output>"
+_TRUSTED_HARNESS_TOOLS = {
+    "load_skill",
+    "read_skill_resource",
+}
+
+
+def spotlight(text: str) -> str:
+    """Wrap opaque tool output as a labelled, JSON-escaped untrusted block."""
+    escaped = json.dumps(text, ensure_ascii=False).replace("/", "\\/")
+    return f"{_SPOTLIGHT_OPEN}\n{escaped}\n{_SPOTLIGHT_CLOSE}"
+
+
+class ToolOutputSecurityMiddleware(FunctionMiddleware):
+    """Spotlight opaque tool results as untrusted data."""
+
+    async def process(
+        self,
+        context: FunctionInvocationContext,
+        call_next: Callable[[], Awaitable[None]],
+    ) -> None:
+        await call_next()
+
+        if (
+            context.function.name in TOOL_REGISTRY
+            or context.function.name in _TRUSTED_HARNESS_TOOLS
+        ):
+            return
+
+        result = context.result
+        if isinstance(result, str) and result:
+            context.result = spotlight(result)
+        elif isinstance(result, list):
+            protected_result = []
+            for item in result:
+                if isinstance(item, Content) and item.type == "text" and item.text:
+                    protected_item = copy(item)
+                    protected_item.text = spotlight(item.text)
+                    protected_result.append(protected_item)
+                elif isinstance(item, Content):
+                    protected_result.append(item)
+                else:
+                    protected_result.append(
+                        spotlight(json.dumps(item, ensure_ascii=False, default=str))
+                    )
+            context.result = protected_result
+        elif isinstance(result, dict):
+            context.result = spotlight(
+                json.dumps(result, ensure_ascii=False, default=str)
+            )