Review MCP Server connection in agents ACP (#563)

* feat(mcp-agent): add rebuild_instruction_templates for post-connect prompt refresh

Add mechanism to rebuild system prompts after MCP servers connect lazily.
This solves the issue where /connect doesn't update {{serverInstructions}}
because the template was already resolved during initialization.

Changes:
- McpAgent: Store original template in _instruction_template
- McpAgent: Add rebuild_instruction_templates() method
- ACPContext: Add invalidate_instruction_cache() to update session caches
- ACPContext: Add set_resolved_instructions() to share cache reference
- AgentACPServer: Wire up resolved_instructions cache to ACPContext
- HuggingFaceAgent: Call rebuild_instruction_templates() after /connect

* refactor(acp): add public API for updating session instructions

Add SlashCommandHandler.update_session_instruction() public method
to avoid ACPContext accessing private _session_instructions directly.
This improves encapsulation and reduces coupling between classes.

* feat(core): add InstructionBuilder for template-based system prompts

Introduce InstructionBuilder class that cleanly separates template storage
from source resolution. The builder produces instruction strings on demand,
supporting both static values and dynamic async resolvers.

Features:
- Static placeholders: {{name}} with .set("name", "value")
- Dynamic resolvers: .set_resolver("name", async_fn) called on each build()
- URL patterns: {{url:https://...}} fetched at build time
- File patterns: {{file:path}} and {{file_silent:path}} relative to workspaceRoot
- Fluent API: builder.set(...).set_resolver(...).build()
- Utility methods: get_placeholders(), get_unresolved_placeholders()

Integration:
- McpAgent now uses InstructionBuilder internally
- Resolvers registered for serverInstructions and agentSkills
- rebuild_instruction_templates() simply calls builder.build()
- Pattern: construct-on-change-and-replace (stateless builder, agent owns result)

Includes 20 unit tests covering all functionality.

* fix(acp): preserve session context when rebuilding instructions

When rebuild_instruction_templates() is called (e.g., after /connect),
the InstructionBuilder was starting from the original template which
still contained {{env}}, {{workspaceRoot}}, etc. These session-level
variables weren't being resolved because the builder only had resolvers
for serverInstructions and agentSkills.

Fix:
- Add McpAgent.set_instruction_context() to set session variables on builder
- Call set_instruction_context() from ACP server during session setup
- Session context (env, workspaceRoot, agentSkills, etc.) is now available
  when rebuilding instructions after /connect

* make instructions visible

* fix(mcp): avoid implicit connect for server instructions

* make server instructions visible (toad xml rendering)

---------

Co-authored-by: Claude <[email protected]>

Author: evalstate

publish/hf-inference-acp/src/hf_inference_acp/agents.py

@@ -497,6 +497,12 @@ async def _send_connect_update(
             self._hf_mcp_connected = True
             await _send_connect_update(title="Connected", status="in_progress")
 
+            # Rebuild system prompt to include fresh server instructions
+            await _send_connect_update(
+                title="Rebuilding system prompt…", status="in_progress"
+            )
+            await self.rebuild_instruction_templates()
+
             # Get available tools
             await _send_connect_update(title="Fetching available tools…", status="in_progress")
             tools_result = await self._aggregator.list_tools()

src/fast_agent/acp/acp_context.py

@@ -147,6 +147,10 @@ def __init__(
         self._progress_manager: "ACPToolProgressManager | None" = None
         self._slash_handler: "SlashCommandHandler | None" = None
 
+        # Reference to session-level resolved instructions cache
+        # (shared with ACPSessionState.resolved_instructions)
+        self._resolved_instructions: dict[str, str] | None = None
+
         # Lock for async operations
         self._lock = asyncio.Lock()
 
@@ -357,6 +361,15 @@ def set_slash_handler(self, handler: "SlashCommandHandler") -> None:
         """Set the slash command handler (called by server)."""
         self._slash_handler = handler
 
+    def set_resolved_instructions(self, resolved_instructions: dict[str, str]) -> None:
+        """
+        Set the reference to resolved instructions cache (called by server).
+
+        This should point to the same dict as ACPSessionState.resolved_instructions
+        so that updates are reflected in both places.
+        """
+        self._resolved_instructions = resolved_instructions
+
     # =========================================================================
     # Slash Command Updates
     # =========================================================================
@@ -416,6 +429,41 @@ async def send_session_update(self, update: Any) -> None:
             update=update,
         )
 
+    async def invalidate_instruction_cache(
+        self, agent_name: str, new_instruction: str | None
+    ) -> None:
+        """
+        Invalidate the session instruction cache for an agent.
+
+        Call this when an agent's system prompt has been rebuilt (e.g., after
+        connecting new MCP servers) to ensure the ACP session uses the fresh
+        instruction on subsequent prompts.
+
+        Args:
+            agent_name: Name of the agent whose instruction was updated
+            new_instruction: The new resolved instruction (or None to remove)
+        """
+        async with self._lock:
+            # Update the session-level resolved instructions cache
+            # (used by _build_session_request_params in AgentACPServer)
+            if self._resolved_instructions is not None:
+                if new_instruction:
+                    self._resolved_instructions[agent_name] = new_instruction
+                elif agent_name in self._resolved_instructions:
+                    del self._resolved_instructions[agent_name]
+
+            # Update the SlashCommandHandler's session instructions cache
+            # (used by /system command)
+            if self._slash_handler:
+                self._slash_handler.update_session_instruction(agent_name, new_instruction)
+
+            logger.info(
+                "Invalidated instruction cache for agent",
+                name="acp_instruction_cache_invalidated",
+                session_id=self._session_id,
+                agent_name=agent_name,
+            )
+
     # =========================================================================
     # Cleanup
     # =========================================================================

src/fast_agent/acp/server/agent_acp_server.py

@@ -748,6 +748,17 @@ async def new_session(
         if resolved_for_session:
             session_state.resolved_instructions = resolved_for_session
 
+        # Set session context on agents that have InstructionBuilder
+        # This ensures {{env}}, {{workspaceRoot}}, etc. are available when rebuilding
+        for agent_name, agent in instance.agents.items():
+            if hasattr(agent, "set_instruction_context"):
+                try:
+                    agent.set_instruction_context(session_context)
+                except Exception as e:
+                    logger.warning(
+                        f"Failed to set instruction context on agent {agent_name}: {e}"
+                    )
+
         # Create slash command handler for this session
         resolved_prompts = session_state.resolved_instructions
 
@@ -784,6 +795,9 @@ async def new_session(
 
             acp_context.set_slash_handler(slash_handler)
 
+            # Share the resolved instructions cache so agents can invalidate it
+            acp_context.set_resolved_instructions(session_state.resolved_instructions)
+
             # Store ACPContext
             session_state.acp_context = acp_context
 

src/fast_agent/acp/slash_commands.py

@@ -174,6 +174,24 @@ def set_current_agent(self, agent_name: str) -> None:
         """
         self.current_agent_name = agent_name
 
+    def update_session_instruction(
+        self, agent_name: str, instruction: str | None
+    ) -> None:
+        """
+        Update the cached session instruction for an agent.
+
+        Call this when an agent's system prompt has been rebuilt (e.g., after
+        connecting new MCP servers) to keep the /system command output current.
+
+        Args:
+            agent_name: Name of the agent whose instruction was updated
+            instruction: The new instruction (or None to remove from cache)
+        """
+        if instruction:
+            self._session_instructions[agent_name] = instruction
+        elif agent_name in self._session_instructions:
+            del self._session_instructions[agent_name]
+
     def _get_current_agent(self) -> AgentProtocol | None:
         """Return the current agent or None if it does not exist."""
         return self.instance.agents.get(self.current_agent_name)

src/fast_agent/agents/mcp_agent.py

@@ -38,6 +38,7 @@
 from fast_agent.agents.tool_agent import ToolAgent
 from fast_agent.constants import HUMAN_INPUT_TOOL_NAME
 from fast_agent.core.exceptions import PromptExitError
+from fast_agent.core.instruction import InstructionBuilder
 from fast_agent.core.logging.logger import get_logger
 from fast_agent.interfaces import FastAgentLLMProtocol
 from fast_agent.mcp.common import (
@@ -102,7 +103,9 @@ def __init__(
             **kwargs,
         )
 
-        self.instruction = self.config.instruction
+        # Store the original template - resolved instruction set after build()
+        self._instruction_template = self.config.instruction
+        self.instruction = self.config.instruction  # Will be replaced by builder output
         self.executor = context.executor if context else None
         self.logger = get_logger(f"{__name__}.{self._name}")
         manifests: list[SkillManifest] = list(getattr(self.config, "skill_manifests", []) or [])
@@ -169,6 +172,13 @@ def __init__(
         if self._shell_runtime_enabled:
             self._shell_runtime.announce()
 
+        # Create instruction builder with dynamic resolvers
+        self._instruction_builder = InstructionBuilder(self._instruction_template or "")
+        self._instruction_builder.set_resolver(
+            "serverInstructions", self._resolve_server_instructions
+        )
+        self._instruction_builder.set_resolver("agentSkills", self._resolve_agent_skills)
+
         # Allow external runtime injection (e.g., for ACP terminal support)
         self._external_runtime = None
 
@@ -268,30 +278,18 @@ async def _apply_instruction_templates(self) -> None:
         Apply template substitution to the instruction, including server instructions.
         This is called during initialization after servers are connected.
         """
-        if not self.instruction:
+        if not self._instruction_builder.template:
             return
 
-        # Gather server instructions if the template includes {{serverInstructions}}
-        if "{{serverInstructions}}" in self.instruction:
-            try:
-                instructions_data = await self._aggregator.get_server_instructions()
-                server_instructions = self._format_server_instructions(instructions_data)
-            except Exception as e:
-                self.logger.warning(f"Failed to get server instructions: {e}")
-                server_instructions = ""
-
-            # Replace the template variable
-            self.instruction = self.instruction.replace(
-                "{{serverInstructions}}", server_instructions
-            )
-
-        skills_placeholder_present = "{{agentSkills}}" in self.instruction
+        # Build the instruction using the InstructionBuilder
+        self.instruction = await self._instruction_builder.build()
 
-        if skills_placeholder_present:
-            agent_skills = format_skills_for_prompt(self._skill_manifests)
-            self.instruction = self.instruction.replace("{{agentSkills}}", agent_skills)
-            self._agent_skills_warning_shown = True
-        elif self._skill_manifests and not self._agent_skills_warning_shown:
+        # Warn if skills configured but placeholder missing
+        if (
+            self._skill_manifests
+            and not self._agent_skills_warning_shown
+            and "{{agentSkills}}" not in self._instruction_builder.template
+        ):
             warning_message = (
                 "Agent skills are configured but the system prompt does not include {{agentSkills}}. "
                 "Skill descriptions will not be added to the system prompt."
@@ -309,6 +307,38 @@ async def _apply_instruction_templates(self) -> None:
 
         self.logger.debug(f"Applied instruction templates for agent {self._name}")
 
+    # ─────────────────────────────────────────────────────────────────────────
+    # Instruction Resolvers (for InstructionBuilder)
+    # ─────────────────────────────────────────────────────────────────────────
+
+    async def _resolve_server_instructions(self) -> str:
+        """Resolver for {{serverInstructions}} placeholder."""
+        try:
+            instructions_data = await self._aggregator.get_server_instructions()
+            return self._format_server_instructions(instructions_data)
+        except Exception as e:
+            self.logger.warning(f"Failed to get server instructions: {e}")
+            return ""
+
+    async def _resolve_agent_skills(self) -> str:
+        """Resolver for {{agentSkills}} placeholder."""
+        self._agent_skills_warning_shown = True
+        return format_skills_for_prompt(self._skill_manifests)
+
+    def set_instruction_context(self, context: dict[str, str]) -> None:
+        """
+        Set session-level context variables on the instruction builder.
+
+        This should be called when an ACP session is established to provide
+        variables like {{env}}, {{workspaceRoot}}, {{agentSkills}} etc. that
+        are resolved per-session.
+
+        Args:
+            context: Dict mapping placeholder names to values (e.g., {"env": "...", "workspaceRoot": "/path"})
+        """
+        self._instruction_builder.set_many(context)
+        self.logger.debug(f"Set instruction context for agent {self._name}: {list(context.keys())}")
+
     def _format_server_instructions(
         self, instructions_data: dict[str, tuple[str | None, list[str]]]
     ) -> str:
@@ -335,16 +365,48 @@ def _format_server_instructions(
             tools_list = ", ".join(prefixed_tools) if prefixed_tools else "No tools available"
 
             formatted_parts.append(
-                f'<mcp-server name="{server_name}">\n'
+                f'<fastagent:mcp-server name="{server_name}">\n'
                 f"<tools>{tools_list}</tools>\n"
                 f"<instructions>\n{instructions}\n</instructions>\n"
-                f"</mcp-server>"
+                f"</fastagent:mcp-server>"
             )
 
         if formatted_parts:
             return "\n\n".join(formatted_parts)
         return ""
 
+    async def rebuild_instruction_templates(self) -> None:
+        """
+        Rebuild instruction from template with fresh source values.
+
+        Call this method after connecting new MCP servers (e.g., via /connect command)
+        to update the system prompt with fresh {{serverInstructions}}.
+
+        The InstructionBuilder re-resolves all dynamic sources (serverInstructions,
+        agentSkills, etc.) each time build() is called.
+        """
+        if not self._instruction_builder.template:
+            return
+
+        # Rebuild using the instruction builder (resolvers are called fresh)
+        self.instruction = await self._instruction_builder.build()
+
+        # Update default request params to match
+        if self._default_request_params:
+            self._default_request_params.systemPrompt = self.instruction
+
+        # Invalidate ACP session caches if running in ACP mode
+        if self.context and hasattr(self.context, "acp") and self.context.acp:
+            try:
+                await self.context.acp.invalidate_instruction_cache(
+                    agent_name=self._name,
+                    new_instruction=self.instruction,
+                )
+            except Exception as e:
+                self.logger.warning(f"Failed to invalidate ACP instruction cache: {e}")
+
+        self.logger.info(f"Rebuilt instruction templates for agent {self._name}")
+
     async def __call__(
         self,
         message: Union[
@@ -430,7 +492,9 @@ def _filter_server_tools(self, tools: list[Tool] | None, namespace: str) -> list
         if namespace not in filters:
             return list(tools)
 
-        filtered = self._filter_server_collections({namespace: tools}, filters, lambda tool: tool.name)
+        filtered = self._filter_server_collections(
+            {namespace: tools}, filters, lambda tool: tool.name
+        )
         return filtered.get(namespace, [])
 
     async def _get_filtered_mcp_tools(self) -> list[Tool]:
@@ -517,7 +581,9 @@ async def call_tool(
                     if name == "read_text_file":
                         return await self._filesystem_runtime.read_text_file(arguments, tool_use_id)
                     elif name == "write_text_file":
-                        return await self._filesystem_runtime.write_text_file(arguments, tool_use_id)
+                        return await self._filesystem_runtime.write_text_file(
+                            arguments, tool_use_id
+                        )
 
         # Fall back to shell runtime
         if self._shell_runtime.tool and name == self._shell_runtime.tool.name:
@@ -905,7 +971,7 @@ async def run_tools(self, request: PromptMessageExtended) -> PromptMessageExtend
                 # Store timing and transport channel info
                 tool_timings[correlation_id] = {
                     "timing_ms": duration_ms,
-                    "transport_channel": getattr(result, "transport_channel", None)
+                    "transport_channel": getattr(result, "transport_channel", None),
                 }
 
                 # Show tool result (like ToolAgent does)
@@ -937,7 +1003,9 @@ async def run_tools(self, request: PromptMessageExtended) -> PromptMessageExtend
                 # Show error result too (no need for skybridge config on errors)
                 self.display.show_tool_result(name=self._name, result=error_result)
 
-        return self._finalize_tool_results(tool_results, tool_timings=tool_timings, tool_loop_error=tool_loop_error)
+        return self._finalize_tool_results(
+            tool_results, tool_timings=tool_timings, tool_loop_error=tool_loop_error
+        )
 
     def _prepare_tool_display(
         self,