diff --git a/mcp_servers.json b/mcp_servers.json index 2d9b84e..3b6d892 100644 --- a/mcp_servers.json +++ b/mcp_servers.json @@ -25,6 +25,7 @@ "git_diff", "list_pending_changes" ] - } + }, + "instructions": "Use this server whenever the user asks about infrastructure, servers, services, networks, documentation, or system inventory. Always validate the repository before making changes. Do not store raw secrets in documentation." } } diff --git a/navi/core/agent.py b/navi/core/agent.py index 70900cc..e95d87a 100644 --- a/navi/core/agent.py +++ b/navi/core/agent.py @@ -212,6 +212,7 @@ profile_registry=profile_registry, memory_store=memory_store, cp_registry=cp_registry, + mcp_manager=mcp_manager, ) self._tool_executor = ToolExecutor(tool_registry) self._planning = PlanningEngine(self._ctx_builder) diff --git a/navi/core/context_builder.py b/navi/core/context_builder.py index 6a42ec9..b841838 100644 --- a/navi/core/context_builder.py +++ b/navi/core/context_builder.py @@ -33,10 +33,12 @@ profile_registry, memory_store: "MemoryStore | None" = None, cp_registry: "ContextProviderRegistry | None" = None, + mcp_manager=None, ) -> None: self._profiles = profile_registry self._memory = memory_store self._cp_registry = cp_registry + self._mcp_manager = mcp_manager self._system_prompt_cache: dict[str, str] = {} def build_system_prompt(self, profile: "AgentProfile") -> str: @@ -189,6 +191,24 @@ lines.append(f"Role: {_role_var.get()}") return Message(role="system", content="\n".join(lines)) + def _mcp_context_msg(self) -> "Message | None": + """Build a system message with MCP server instructions. + + Combines server-provided instructions (from MCP initialize handshake) + with overlay instructions from ``mcp_servers.json``. + """ + if not self._mcp_manager: + return None + instructions = self._mcp_manager.get_instructions() + if not instructions: + return None + lines = ["[MCP servers — external knowledge sources]"] + for name, text in instructions.items(): + lines.append(f"") + lines.append(f"## {name}") + lines.append(text) + return Message(role="system", content="\n".join(lines)) + def build( self, session_context: list[Message], @@ -224,6 +244,11 @@ if policy: result.append(policy) + # Inject MCP server instructions into context + mcp_msg = self._mcp_context_msg() + if mcp_msg: + result.append(mcp_msg) + if extra_system: result.extend(extra_system) result.extend(conv) diff --git a/navi/mcp/client.py b/navi/mcp/client.py index 02977b9..78191c8 100644 --- a/navi/mcp/client.py +++ b/navi/mcp/client.py @@ -30,11 +30,17 @@ self._session: ClientSession | None = None self._exit_stack = AsyncExitStack() self._connected = False + self._instructions: str | None = None @property def connected(self) -> bool: return self._connected and self._session is not None + @property + def instructions(self) -> str | None: + """Server-provided instructions from MCP initialize handshake.""" + return self._instructions + async def connect(self) -> None: """Open transport, initialise session, and store it.""" if self._connected: @@ -69,10 +75,15 @@ session = await self._exit_stack.enter_async_context( ClientSession(read_stream, write_stream) ) - await session.initialize() + init_result = await session.initialize() + self._instructions = init_result.instructions if hasattr(init_result, "instructions") else None self._session = session self._connected = True - logger.info("MCP server %r connected (%s)", self.name, self.config.transport) + logger.info( + "MCP server %r connected (%s)", + self.name, + self.config.transport, + ) except Exception: await self._cleanup() raise @@ -90,6 +101,7 @@ pass finally: self._session = None + self._instructions = None self._connected = False self._exit_stack = AsyncExitStack() diff --git a/navi/mcp/config.py b/navi/mcp/config.py index 74d337d..239ab73 100644 --- a/navi/mcp/config.py +++ b/navi/mcp/config.py @@ -26,6 +26,10 @@ # Profiles reference groups by name instead of listing individual tools. groups: dict[str, list[str]] = Field(default_factory=dict) + # Overlay instructions injected into Navi's system prompt alongside the + # instructions provided by the MCP server itself during the initialize handshake. + instructions: str | None = None + @property def is_stdio(self) -> bool: return self.transport == "stdio" diff --git a/navi/mcp/manager.py b/navi/mcp/manager.py index f5b967c..6ab1603 100644 --- a/navi/mcp/manager.py +++ b/navi/mcp/manager.py @@ -94,6 +94,28 @@ return [] return list(cfg.groups.get(group_name, [])) + def get_instructions(self) -> dict[str, str]: + """Return combined instructions for every connected server. + + Server-provided instructions (from MCP initialize handshake) are merged + with the overlay ``instructions`` field from ``mcp_servers.json``. + If a server is disconnected, only the config overlay is returned. + """ + configs = load_mcp_servers(self.config_path) + out: dict[str, str] = {} + for name, client in self._clients.items(): + parts: list[str] = [] + if client.instructions: + parts.append(client.instructions) + cfg = configs.get(name) + if cfg and cfg.instructions: + if parts: + parts.append("") + parts.append(cfg.instructions) + if parts: + out[name] = "\n".join(parts) + return out + async def call_tool(self, server_name: str, tool_name: str, arguments: dict[str, Any] | None = None) -> str: """Proxy a tool call to the named server.""" client = self._clients.get(server_name)