feat(mcp): expose MCP resources and prompts as read-only tools
Add MCPResourceWrapper and MCPPromptWrapper classes that expose MCP server resources and prompts as nanobot tools. Resources are read-only tools that fetch content by URI, and prompts are read-only tools that return filled prompt templates with optional arguments. - MCPResourceWrapper: reads resource content (text and binary) via URI - MCPPromptWrapper: gets prompt templates with typed arguments - Both handle timeouts, cancellation, and MCP SDK 1.x error types - Resources and prompts are registered during server connection - Gracefully handles servers that don't support resources/prompts
This commit is contained in:
Tim O'Brien
Xubin Ren
@@ -7,7 +7,12 @@ from types import ModuleType, SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from nanobot.agent.tools.mcp import MCPToolWrapper, connect_mcp_servers
|
||||
from nanobot.agent.tools.mcp import (
|
||||
MCPResourceWrapper,
|
||||
MCPPromptWrapper,
|
||||
MCPToolWrapper,
|
||||
connect_mcp_servers,
|
||||
)
|
||||
from nanobot.agent.tools.registry import ToolRegistry
|
||||
from nanobot.config.schema import MCPServerConfig
|
||||
|
||||
@@ -17,6 +22,16 @@ class _FakeTextContent:
|
||||
self.text = text
|
||||
|
||||
|
||||
class _FakeTextResourceContents:
|
||||
def __init__(self, text: str) -> None:
|
||||
self.text = text
|
||||
|
||||
|
||||
class _FakeBlobResourceContents:
|
||||
def __init__(self, blob: bytes) -> None:
|
||||
self.blob = blob
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_mcp_runtime() -> dict[str, object | None]:
|
||||
return {"session": None}
|
||||
@@ -27,7 +42,11 @@ def _fake_mcp_module(
|
||||
monkeypatch: pytest.MonkeyPatch, fake_mcp_runtime: dict[str, object | None]
|
||||
) -> None:
|
||||
mod = ModuleType("mcp")
|
||||
mod.types = SimpleNamespace(TextContent=_FakeTextContent)
|
||||
mod.types = SimpleNamespace(
|
||||
TextContent=_FakeTextContent,
|
||||
TextResourceContents=_FakeTextResourceContents,
|
||||
BlobResourceContents=_FakeBlobResourceContents,
|
||||
)
|
||||
|
||||
class _FakeStdioServerParameters:
|
||||
def __init__(self, command: str, args: list[str], env: dict | None = None) -> None:
|
||||
@@ -343,3 +362,237 @@ async def test_connect_mcp_servers_enabled_tools_warns_on_unknown_entries(
|
||||
assert "enabledTools entries not found: unknown" in warnings[-1]
|
||||
assert "Available raw names: demo" in warnings[-1]
|
||||
assert "Available wrapped names: mcp_test_demo" in warnings[-1]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MCPResourceWrapper tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_resource_def(
|
||||
name: str = "myres",
|
||||
uri: str = "file:///tmp/data.txt",
|
||||
description: str = "A test resource",
|
||||
) -> SimpleNamespace:
|
||||
return SimpleNamespace(name=name, uri=uri, description=description)
|
||||
|
||||
|
||||
def _make_resource_wrapper(
|
||||
session: object, *, timeout: float = 0.1
|
||||
) -> MCPResourceWrapper:
|
||||
return MCPResourceWrapper(session, "srv", _make_resource_def(), resource_timeout=timeout)
|
||||
|
||||
|
||||
def test_resource_wrapper_properties() -> None:
|
||||
wrapper = MCPResourceWrapper(None, "myserver", _make_resource_def())
|
||||
assert wrapper.name == "mcp_myserver_resource_myres"
|
||||
assert "[MCP Resource]" in wrapper.description
|
||||
assert "A test resource" in wrapper.description
|
||||
assert "file:///tmp/data.txt" in wrapper.description
|
||||
assert wrapper.parameters == {"type": "object", "properties": {}, "required": []}
|
||||
assert wrapper.read_only is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resource_wrapper_execute_returns_text() -> None:
|
||||
async def read_resource(uri: str) -> object:
|
||||
assert uri == "file:///tmp/data.txt"
|
||||
return SimpleNamespace(
|
||||
contents=[_FakeTextResourceContents("line1"), _FakeTextResourceContents("line2")]
|
||||
)
|
||||
|
||||
wrapper = _make_resource_wrapper(SimpleNamespace(read_resource=read_resource))
|
||||
result = await wrapper.execute()
|
||||
assert result == "line1\nline2"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resource_wrapper_execute_handles_blob() -> None:
|
||||
async def read_resource(uri: str) -> object:
|
||||
return SimpleNamespace(contents=[_FakeBlobResourceContents(b"\x00\x01\x02")])
|
||||
|
||||
wrapper = _make_resource_wrapper(SimpleNamespace(read_resource=read_resource))
|
||||
result = await wrapper.execute()
|
||||
assert "[Binary resource: 3 bytes]" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resource_wrapper_execute_handles_timeout() -> None:
|
||||
async def read_resource(uri: str) -> object:
|
||||
await asyncio.sleep(1)
|
||||
return SimpleNamespace(contents=[])
|
||||
|
||||
wrapper = _make_resource_wrapper(
|
||||
SimpleNamespace(read_resource=read_resource), timeout=0.01
|
||||
)
|
||||
result = await wrapper.execute()
|
||||
assert result == "(MCP resource read timed out after 0.01s)"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resource_wrapper_execute_handles_error() -> None:
|
||||
async def read_resource(uri: str) -> object:
|
||||
raise RuntimeError("boom")
|
||||
|
||||
wrapper = _make_resource_wrapper(SimpleNamespace(read_resource=read_resource))
|
||||
result = await wrapper.execute()
|
||||
assert result == "(MCP resource read failed: RuntimeError)"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MCPPromptWrapper tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_prompt_def(
|
||||
name: str = "myprompt",
|
||||
description: str = "A test prompt",
|
||||
arguments: list | None = None,
|
||||
) -> SimpleNamespace:
|
||||
return SimpleNamespace(name=name, description=description, arguments=arguments)
|
||||
|
||||
|
||||
def _make_prompt_wrapper(
|
||||
session: object, *, timeout: float = 0.1
|
||||
) -> MCPPromptWrapper:
|
||||
return MCPPromptWrapper(
|
||||
session, "srv", _make_prompt_def(), prompt_timeout=timeout
|
||||
)
|
||||
|
||||
|
||||
def test_prompt_wrapper_properties() -> None:
|
||||
arg1 = SimpleNamespace(name="topic", required=True)
|
||||
arg2 = SimpleNamespace(name="style", required=False)
|
||||
wrapper = MCPPromptWrapper(
|
||||
None, "myserver", _make_prompt_def(arguments=[arg1, arg2])
|
||||
)
|
||||
assert wrapper.name == "mcp_myserver_prompt_myprompt"
|
||||
assert "[MCP Prompt]" in wrapper.description
|
||||
assert "A test prompt" in wrapper.description
|
||||
assert "workflow guide" in wrapper.description
|
||||
assert wrapper.parameters["properties"]["topic"] == {"type": "string"}
|
||||
assert wrapper.parameters["properties"]["style"] == {"type": "string"}
|
||||
assert wrapper.parameters["required"] == ["topic"]
|
||||
assert wrapper.read_only is True
|
||||
|
||||
|
||||
def test_prompt_wrapper_no_arguments() -> None:
|
||||
wrapper = MCPPromptWrapper(None, "myserver", _make_prompt_def())
|
||||
assert wrapper.parameters == {"type": "object", "properties": {}, "required": []}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prompt_wrapper_execute_returns_text() -> None:
|
||||
async def get_prompt(name: str, arguments: dict | None = None) -> object:
|
||||
assert name == "myprompt"
|
||||
msg1 = SimpleNamespace(
|
||||
role="user",
|
||||
content=[_FakeTextContent("You are an expert on {{topic}}.")],
|
||||
)
|
||||
msg2 = SimpleNamespace(
|
||||
role="assistant",
|
||||
content=[_FakeTextContent("Understood. Ask me anything.")],
|
||||
)
|
||||
return SimpleNamespace(messages=[msg1, msg2])
|
||||
|
||||
wrapper = _make_prompt_wrapper(SimpleNamespace(get_prompt=get_prompt))
|
||||
result = await wrapper.execute(topic="AI")
|
||||
assert "You are an expert on {{topic}}." in result
|
||||
assert "Understood. Ask me anything." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prompt_wrapper_execute_handles_timeout() -> None:
|
||||
async def get_prompt(name: str, arguments: dict | None = None) -> object:
|
||||
await asyncio.sleep(1)
|
||||
return SimpleNamespace(messages=[])
|
||||
|
||||
wrapper = _make_prompt_wrapper(
|
||||
SimpleNamespace(get_prompt=get_prompt), timeout=0.01
|
||||
)
|
||||
result = await wrapper.execute()
|
||||
assert result == "(MCP prompt call timed out after 0.01s)"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prompt_wrapper_execute_handles_error() -> None:
|
||||
async def get_prompt(name: str, arguments: dict | None = None) -> object:
|
||||
raise RuntimeError("boom")
|
||||
|
||||
wrapper = _make_prompt_wrapper(SimpleNamespace(get_prompt=get_prompt))
|
||||
result = await wrapper.execute()
|
||||
assert result == "(MCP prompt call failed: RuntimeError)"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# connect_mcp_servers: resources + prompts integration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_fake_session_with_capabilities(
|
||||
tool_names: list[str],
|
||||
resource_names: list[str] | None = None,
|
||||
prompt_names: list[str] | None = None,
|
||||
) -> SimpleNamespace:
|
||||
async def initialize() -> None:
|
||||
return None
|
||||
|
||||
async def list_tools() -> SimpleNamespace:
|
||||
return SimpleNamespace(tools=[_make_tool_def(name) for name in tool_names])
|
||||
|
||||
async def list_resources() -> SimpleNamespace:
|
||||
resources = []
|
||||
for rname in resource_names or []:
|
||||
resources.append(
|
||||
SimpleNamespace(
|
||||
name=rname,
|
||||
uri=f"file:///{rname}",
|
||||
description=f"{rname} resource",
|
||||
)
|
||||
)
|
||||
return SimpleNamespace(resources=resources)
|
||||
|
||||
async def list_prompts() -> SimpleNamespace:
|
||||
prompts = []
|
||||
for pname in prompt_names or []:
|
||||
prompts.append(
|
||||
SimpleNamespace(
|
||||
name=pname,
|
||||
description=f"{pname} prompt",
|
||||
arguments=None,
|
||||
)
|
||||
)
|
||||
return SimpleNamespace(prompts=prompts)
|
||||
|
||||
return SimpleNamespace(
|
||||
initialize=initialize,
|
||||
list_tools=list_tools,
|
||||
list_resources=list_resources,
|
||||
list_prompts=list_prompts,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_connect_registers_resources_and_prompts(
|
||||
fake_mcp_runtime: dict[str, object | None],
|
||||
) -> None:
|
||||
fake_mcp_runtime["session"] = _make_fake_session_with_capabilities(
|
||||
tool_names=["tool_a"],
|
||||
resource_names=["res_b"],
|
||||
prompt_names=["prompt_c"],
|
||||
)
|
||||
registry = ToolRegistry()
|
||||
stack = AsyncExitStack()
|
||||
await stack.__aenter__()
|
||||
try:
|
||||
await connect_mcp_servers(
|
||||
{"test": MCPServerConfig(command="fake")},
|
||||
registry,
|
||||
stack,
|
||||
)
|
||||
finally:
|
||||
await stack.aclose()
|
||||
|
||||
assert "mcp_test_tool_a" in registry.tool_names
|
||||
assert "mcp_test_resource_res_b" in registry.tool_names
|
||||
assert "mcp_test_prompt_prompt_c" in registry.tool_names
|
||||
|
||||
Reference in New Issue
Block a user