142 lines
4.4 KiB
Python
142 lines
4.4 KiB
Python
"""Tests for MCPToolbox."""
|
|
|
|
import pytest
|
|
|
|
from mcp_api import MCPTool, MCPToolbox
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Discovery
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_discovers_all_tools(toolbox: MCPToolbox) -> None:
|
|
names = {t.name for t in toolbox.tools}
|
|
assert names == {"echo", "reverse", "add", "multiply"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_has_description(toolbox: MCPToolbox) -> None:
|
|
for tool in toolbox.tools:
|
|
assert tool.description, f"{tool.name} has no description"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_has_input_schema(toolbox: MCPToolbox) -> None:
|
|
for tool in toolbox.tools:
|
|
assert isinstance(tool.input_schema, dict)
|
|
assert tool.input_schema.get("type") == "object"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool access
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_getitem(toolbox: MCPToolbox) -> None:
|
|
assert isinstance(toolbox["echo"], MCPTool)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flat_attribute_access(toolbox: MCPToolbox) -> None:
|
|
# "echo" is also a server name → _ServerNamespace takes priority
|
|
# Use a tool name that doesn't collide with a server name for flat access
|
|
assert isinstance(toolbox.reverse, MCPTool)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_server_namespace_attribute_access(toolbox: MCPToolbox) -> None:
|
|
assert isinstance(toolbox.echo.echo, MCPTool)
|
|
assert isinstance(toolbox.math.add, MCPTool)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_attribute_raises(toolbox: MCPToolbox) -> None:
|
|
with pytest.raises(AttributeError):
|
|
_ = toolbox.nonexistent
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool calls — direct
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_tool_echo(toolbox: MCPToolbox) -> None:
|
|
result = await toolbox.call_tool("echo", {"message": "hello"})
|
|
assert result == "hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_tool_reverse(toolbox: MCPToolbox) -> None:
|
|
result = await toolbox.call_tool("reverse", {"text": "abc"})
|
|
assert result == "cba"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_tool_add(toolbox: MCPToolbox) -> None:
|
|
result = await toolbox.call_tool("add", {"a": 3, "b": 4})
|
|
assert result == "7"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_tool_multiply(toolbox: MCPToolbox) -> None:
|
|
result = await toolbox.call_tool("multiply", {"a": 6, "b": 7})
|
|
assert result == "42"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool calls — via callable / attribute access
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callable_tool(toolbox: MCPToolbox) -> None:
|
|
result = await toolbox["echo"](message="world")
|
|
assert result == "world"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_server_namespace_call(toolbox: MCPToolbox) -> None:
|
|
result = await toolbox.math.add(a=10, b=5)
|
|
assert result == "15"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Anthropic format
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_as_anthropic_tools_structure(toolbox: MCPToolbox) -> None:
|
|
defs = toolbox.as_anthropic_tools()
|
|
assert len(defs) == 4
|
|
for d in defs:
|
|
assert {"name", "description", "input_schema"} == d.keys()
|
|
assert isinstance(d["input_schema"], dict)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_anthropic_tool_use_success(toolbox: MCPToolbox) -> None:
|
|
class FakeBlock:
|
|
type = "tool_use"
|
|
id = "tu_123"
|
|
name = "echo"
|
|
input = {"message": "ping"}
|
|
|
|
result = await toolbox.handle_anthropic_tool_use(FakeBlock())
|
|
assert result["type"] == "tool_result"
|
|
assert result["tool_use_id"] == "tu_123"
|
|
assert result["content"] == "ping"
|
|
assert result["is_error"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_anthropic_tool_use_error(toolbox: MCPToolbox) -> None:
|
|
class FakeBlock:
|
|
type = "tool_use"
|
|
id = "tu_456"
|
|
name = "nonexistent_tool"
|
|
input = {}
|
|
|
|
result = await toolbox.handle_anthropic_tool_use(FakeBlock())
|
|
assert result["is_error"] is True
|
|
assert result["tool_use_id"] == "tu_456"
|