跳转到主要内容

直接对工具进行单元测试

工具本质上就是函数。无需服务器即可测试它们:
def test_add():
    assert add(2, 3) == 5

async def test_async_tool():
    result = await fetch_user("123")
    assert result["id"] == "123"

使用上下文的工具

调用 get_context() 的工具需要有一个正在进行的请求。通过连接真实服务器的集成测试来验证它们,或者调整你的 tool 设计,使依赖上下文的部分便于进行 mock:
@tool(description="Process with logging")
async def process(data: str) -> dict:
    ctx = get_context()
    await ctx.info("Processing")
    return do_work(data)  # 单独测试 do_work

def test_do_work():
    assert do_work("input") == {"result": "output"}

使用 MCPClient 进行集成测试

对整个服务器进行测试:
import pytest
from dedalus_mcp import MCPServer, tool
from dedalus_mcp.client import MCPClient

@tool(description="Add")
def add(a: int, b: int) -> int:
    return a + b

@pytest.fixture
async def server():
    server = MCPServer("test")
    server.collect(add)
    # 在后台启动服务器
    task = asyncio.create_task(server.serve())
    await asyncio.sleep(0.1)  # 等待启动
    yield server
    task.cancel()

@pytest.fixture
async def client(server):
    client = await MCPClient.connect("http://127.0.0.1:8000/mcp")
    yield client
    await client.close()

async def test_call_tool(client):
    result = await client.call_tool("add", {"a": 2, "b": 3})
    assert result.content[0].text == "5"

测试注册

验证工具是否正确注册:
def test_registration():
    server = MCPServer("test")
    server.collect(add, multiply)
    
    names = list(server.tool_names)
    
    assert "add" in names
    assert "multiply" in names

隔离

由于装饰器不会在导入时绑定到服务器,每个测试都能获得干净的初始状态:
def test_a():
    server = MCPServer("test-a")
    server.collect(tool_a)
    # 无需清理

def test_b():
    server = MCPServer("test-b")
    server.collect(tool_b)
    # 完全独立