cli: fold MCP initialize instructions into agent prompt and /mcp

This commit is contained in:
2026-09-08 13:09:32 +08:00
parent 5f28a49bc1
commit 4f142f1db9
3 changed files with 133 additions and 0 deletions
+11
View File
@@ -1147,6 +1147,9 @@ def plugins_slash_cmd(state: ReplState, action: str | None, name: str | None) ->
raise click.UsageError(str(exc)) from exc raise click.UsageError(str(exc)) from exc
_MCP_PREVIEW_MAX_CHARS = 120
def _print_mcp_status(state: ReplState) -> None: def _print_mcp_status(state: ReplState) -> None:
manager = state.mcp_manager manager = state.mcp_manager
if manager is None: if manager is None:
@@ -1162,6 +1165,14 @@ def _print_mcp_status(state: ReplState) -> None:
(conn for conn in manager.connections() if conn.name == name), (conn for conn in manager.connections() if conn.name == name),
None, None,
) )
if connection is not None and connection.instructions:
preview = " ".join(connection.instructions.split())
if len(preview) > _MCP_PREVIEW_MAX_CHARS:
preview = f"{preview[: _MCP_PREVIEW_MAX_CHARS - 3]}..."
click.secho(
f" instructions ({len(connection.instructions)} chars): {preview}",
fg="bright_black",
)
if connection is not None and connection.error is not None: if connection is not None and connection.error is not None:
for line in connection.stderr_tail()[-3:]: for line in connection.stderr_tail()[-3:]:
click.secho(f" {line}", fg="bright_black") click.secho(f" {line}", fg="bright_black")
+14
View File
@@ -237,14 +237,28 @@ class ReplState:
session_state=self.session_state, session_state=self.session_state,
) )
def _mcp_instructions_text(self) -> str:
"""Compose connected MCP servers' initialize ``instructions`` into one block.
Only meaningful when tools are on (MCP tools are registered under the
same condition); the text is attributed per server so the model can
weigh its provenance. Gated by ``[agent] mcp_instructions`` in callers.
"""
if self.mcp_manager is None or not self.tools_enabled:
return ""
parts = [f"MCP server {name!r} instructions:\n{text}" for name, text in self.mcp_manager.instructions()]
return "\n\n".join(parts)
def _make_agent(self) -> ChatAgent: def _make_agent(self) -> ChatAgent:
from plyngent.cli.limits import prompt_continue_limit_async from plyngent.cli.limits import prompt_continue_limit_async
from plyngent.config import compose_agent_system_content from plyngent.config import compose_agent_system_content
agent_cfg = self.config.agent_config agent_cfg = self.config.agent_config
mcp_text = self._mcp_instructions_text() if agent_cfg.mcp_instructions else ""
system_prompt = compose_agent_system_content( system_prompt = compose_agent_system_content(
agent_cfg.system_prompt, agent_cfg.system_prompt,
agent_cfg.tool_directives, agent_cfg.tool_directives,
mcp_text,
) )
on_limit = prompt_continue_limit_async if self.interactive_limits else None on_limit = prompt_continue_limit_async if self.interactive_limits else None
peak = 0 peak = 0
+108
View File
@@ -120,3 +120,111 @@ async def test_mcp_tools_registered_on_tools_rebuild(state: ReplState) -> None:
assert out == "echo: hi" assert out == "echo: hi"
finally: finally:
await manager.aclose() await manager.aclose()
async def test_slash_mcp_list_shows_instructions(
state: ReplState,
capsys: pytest.CaptureFixture[str],
) -> None:
guidance = "Prefer `echo` for repeating text back."
manager = McpManager(
McpConfig(
servers={
"fake": _server_config(env={"PLYNGENT_MCP_FAKE_INSTRUCTIONS": guidance}),
},
disable=[],
)
)
await manager.ensure_started()
state.mcp_manager = manager
try:
assert await handle_slash(state, "/mcp list") is True
out = capsys.readouterr().out
assert f"instructions ({len(guidance)} chars)" in out
assert guidance in out
finally:
await manager.aclose()
async def _make_tools_state(
tmp_path: Path,
*,
agent_toml: str,
manager: McpManager,
) -> tuple[ReplState, MemoryStore]:
"""ReplState with tools on + a started MCP manager (agent built from TOML)."""
memory = await MemoryStore.open(DatabaseConfig())
provider = OpenAIProvider(access_key_or_token="sk-test")
document = tomlkit.parse(agent_toml) if agent_toml else tomlkit.document()
config = ConfigStore(path=tmp_path / "plyngent.toml", document=document)
config.providers = {"local": provider}
st = ReplState(
config=config,
memory=memory,
workspace=tmp_path,
provider_name="local",
provider=provider,
model="gpt-test",
tools_enabled=True,
mcp_manager=manager,
)
return st, memory
async def test_mcp_instructions_folded_into_agent_system_prompt(tmp_path: Path) -> None:
from plyngent.tools.catalog import catalog_scope
guidance = "Prefer `echo` for repeating text back to the user."
manager = McpManager(
McpConfig(
servers={
"fake": _server_config(env={"PLYNGENT_MCP_FAKE_INSTRUCTIONS": guidance}),
},
disable=[],
)
)
await manager.ensure_started()
try:
with catalog_scope(empty=True):
state, memory = await _make_tools_state(tmp_path, agent_toml="", manager=manager)
try:
prompt = state.agent.system_prompt
assert prompt is not None
assert "MCP server 'fake' instructions:" in prompt
assert guidance in prompt
finally:
await memory.close()
finally:
await manager.aclose()
async def test_mcp_instructions_flag_off_skips_server_text(tmp_path: Path) -> None:
from plyngent.tools.catalog import catalog_scope
guidance = "Prefer `echo` for repeating text back to the user."
manager = McpManager(
McpConfig(
servers={
"fake": _server_config(env={"PLYNGENT_MCP_FAKE_INSTRUCTIONS": guidance}),
},
disable=[],
)
)
await manager.ensure_started()
try:
with catalog_scope(empty=True):
state, memory = await _make_tools_state(
tmp_path,
agent_toml="[agent]\nmcp_instructions = false\n",
manager=manager,
)
try:
prompt = state.agent.system_prompt
assert prompt is not None
assert guidance not in prompt
assert "MCP server" not in prompt
assert "### Workspace" in prompt # host tool playbook still composed
finally:
await memory.close()
finally:
await manager.aclose()