mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
NCBM
e9ee71e853
New `view_image` reads a local picture (JPEG, PNG, GIF, WebP, sniffed from its bytes) and hands it to the model as an image part, with an optional `detail` level — `low` measured 3.4x cheaper on a 1200x900 picture. `read_file` now points at it instead of returning decode noise. `fetch_local` returns an image response as the picture itself, and MCP image content blocks become image parts instead of `[image: ...]` placeholders, so a screenshot server finally reaches the model.
231 lines
7.5 KiB
Python
231 lines
7.5 KiB
Python
"""CLI slash ``/mcp`` command and MCP wiring in ``ReplState``."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
import pytest
|
|
import tomlkit
|
|
|
|
from plyngent.cli.slash import handle_slash
|
|
from plyngent.cli.state import ReplState
|
|
from plyngent.config.models import DatabaseConfig, McpConfig, McpServerConfig, OpenAIProvider
|
|
from plyngent.config.store import ConfigStore
|
|
from plyngent.memory import MemoryStore
|
|
from plyngent.runtime.mcp_client import McpManager
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import AsyncIterator
|
|
|
|
_FAKE_SERVER = Path(__file__).parent.parent / "test_runtime" / "mcp_fake_server.py"
|
|
|
|
|
|
def _server_config(**overrides: object) -> McpServerConfig:
|
|
base: dict[str, object] = {
|
|
"command": sys.executable,
|
|
"args": [str(_FAKE_SERVER)],
|
|
"timeout": 5.0,
|
|
}
|
|
return McpServerConfig(**{**base, **overrides}) # type: ignore[arg-type]
|
|
|
|
|
|
@pytest.fixture
|
|
async def state(tmp_path: Path) -> AsyncIterator[ReplState]:
|
|
memory = await MemoryStore.open(DatabaseConfig())
|
|
provider = OpenAIProvider(access_key_or_token="sk-test")
|
|
config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document())
|
|
config.providers = {"local": provider}
|
|
st = ReplState(
|
|
config=config,
|
|
memory=memory,
|
|
workspace=tmp_path,
|
|
provider_name="local",
|
|
provider=provider,
|
|
model="gpt-test",
|
|
tools_enabled=False,
|
|
)
|
|
await st.new_session("t")
|
|
yield st
|
|
await memory.close()
|
|
|
|
|
|
async def test_slash_mcp_list_without_manager(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
|
|
assert await handle_slash(state, "/mcp") is True
|
|
assert "no servers configured" in capsys.readouterr().out
|
|
assert await handle_slash(state, "/mcp list") is True
|
|
assert "no servers configured" in capsys.readouterr().out
|
|
|
|
|
|
async def test_slash_mcp_list_statuses(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
|
|
manager = McpManager(McpConfig(servers={"fake": _server_config()}, disable=[]))
|
|
await manager.ensure_started()
|
|
state.mcp_manager = manager
|
|
try:
|
|
assert await handle_slash(state, "/mcp list") is True
|
|
out = capsys.readouterr().out
|
|
assert "fake" in out
|
|
assert "connected" in out
|
|
assert "tools=4" in out
|
|
finally:
|
|
await manager.aclose()
|
|
|
|
|
|
async def test_slash_mcp_reconnect_connects_new_config(
|
|
state: ReplState,
|
|
capsys: pytest.CaptureFixture[str],
|
|
tmp_path: Path,
|
|
) -> None:
|
|
# The fixture's config file does not exist yet; write one with an MCP server.
|
|
payload = (
|
|
"[mcp.servers.fake]\n"
|
|
f"command = {json.dumps(sys.executable)}\n"
|
|
f"args = [{json.dumps(str(_FAKE_SERVER))}]\n"
|
|
"timeout = 5.0\n"
|
|
)
|
|
_ = (tmp_path / "plyngent.toml").write_text(payload, encoding="utf-8")
|
|
assert await handle_slash(state, "/mcp reconnect") is True
|
|
try:
|
|
assert state.mcp_manager is not None
|
|
statuses = state.mcp_manager.statuses()
|
|
assert [name for name, _, _ in statuses] == ["fake"]
|
|
assert statuses[0][1] == "connected"
|
|
assert statuses[0][2] == 4
|
|
out = capsys.readouterr().out
|
|
assert "tools=off" in out
|
|
assert "fake" in out
|
|
finally:
|
|
if state.mcp_manager is not None:
|
|
await state.mcp_manager.aclose()
|
|
|
|
|
|
async def test_mcp_tools_registered_on_tools_rebuild(state: ReplState) -> None:
|
|
manager = McpManager(McpConfig(servers={"fake": _server_config()}, disable=[]))
|
|
await manager.ensure_started()
|
|
state.mcp_manager = manager
|
|
try:
|
|
from plyngent.tools.catalog import catalog_scope
|
|
|
|
state.tools_enabled = True
|
|
with catalog_scope(empty=True):
|
|
state.rebuild_client()
|
|
registry = state.agent.tools
|
|
assert registry is not None
|
|
assert registry.get("mcp__fake__echo") is not None
|
|
assert registry.get("mcp__fake__fail") is not None
|
|
# Handler proxies to the live server subprocess.
|
|
out = await registry.execute("mcp__fake__echo", '{"text": "hi"}')
|
|
assert out == "echo: hi"
|
|
finally:
|
|
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()
|