跳转到主要内容
工具让智能体可以调用你的 Python 函数。装饰、注册、提供服务。 工具是核心构件，使 MCP Client 可以通过模型上下文协议 (MCP) 调用你的 Python 函数：
  1. Client 通过 tools/list 发现工具（每个工具都包含 inputSchema 和可选的 outputSchema）。
  2. Client 通过 tools/call 调用某个 tool，并使用与 schema 匹配的 arguments。
  3. 服务器执行你注册的可调用对象。
  4. 服务器根据 MCP 规范返回包含 content（以及可选的 structuredContent）的 CallToolResult。
## 基础的 tool
from dedalus_mcp import MCPServer, tool

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

server = MCPServer("math")
server.collect(add)
description 告诉 LLM 这个 tool 的功能。类型注解会被转换为 JSON Schema。

​
异步工具

import anyio
from dedalus_mcp import tool

@tool(description="Fetch user data (simulated I/O)")
async def get_user(user_id: str) -> dict:
    await anyio.sleep(0.1)
    return {"user_id": user_id, "status": "ok"}
对于 I/O，优先使用 async。 重要提示：在 Dedalus 模型上下文协议 (MCP) 中，同步工具会内联运行（不会自动放到线程池中）。如果你需要为阻塞型工作提供并发能力，请使用 async def 并显式地将阻塞任务进行异步 offload。

​
类型推断

类型注解会自动转换为 JSON Schema： 
from typing import Literal
from pydantic import BaseModel
from dedalus_mcp import tool

class SearchFilters(BaseModel):
    category: str | None = None
    min_price: float = 0.0

@tool(description="Search products")
def search(
    query: str,
    limit: int = 10,
    sort: Literal["relevance", "price", "date"] = "relevance",
    filters: SearchFilters | None = None,
) -> list[dict]:
    return [{"query": query, "limit": limit, "sort": sort, "filters": filters.model_dump() if filters else None}]
支持：基础类型、list、dict、Literal、Enum、可选/联合类型、Pydantic 模型、数据类、嵌套模型。 必填参数没有默认值，可选参数则有默认值。

​
装饰器选项

from dedalus_mcp import tool

@tool(
    name="find_products",           # 覆盖 tool 名称
    description="Search catalog",   # Tool 描述
    tags={"search", "catalog"},     # 用于筛选/元数据
)
def search_products_impl(query: str) -> list[dict]:
    return [{"id": "p_1", "name": "Widget", "query": query}]

​
结构化返回值

返回可序列化为 JSON 的值： 
from dedalus_mcp import tool

@tool(description="Analyze text")
def analyze(text: str) -> dict:
    return {"word_count": len(text.split()), "char_count": len(text)}
如需进行显式控制，请返回 CallToolResult： 
from dedalus_mcp import tool
from dedalus_mcp.types import CallToolResult, TextContent

@tool(description="Custom result")
def custom() -> CallToolResult:
    return CallToolResult(
        content=[TextContent(type="text", text="Custom message")],
        isError=False,
    )

​
上下文访问

通过 get_context() 进行日志记录与进度跟踪： 
import anyio
from dedalus_mcp import tool, get_context

@tool(description="Process files with progress reporting")
async def process_files(paths: list[str]) -> dict:
    ctx = get_context()
    await ctx.info("Starting", data={"count": len(paths)})

    processed = 0
    try:
        async with ctx.progress(total=len(paths)) as tracker:
            for path in paths:
                # 模拟工作；取消信号以任务取消的形式传递
                await anyio.sleep(0.01)
                processed += 1
                await tracker.advance(1)
    except anyio.get_cancelled_exc_class():
        await ctx.warning("Cancelled", data={"processed": processed})
        raise

    return {"processed": processed}

​
允许列表

限制可见工具： 
from dedalus_mcp import MCPServer, tool

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

@tool(description="Multiply")
def multiply(a: int, b: int) -> int:
    return a * b

server = MCPServer("gated")
server.collect(add, multiply)
server.allow_tools({"add"})
调用隐藏的 tool 会返回一个错误 CallToolResult，表明该 tool 当前不可用。

​
错误处理

按常规方式抛出异常： 
from dedalus_mcp import tool

@tool(description="Divide")
def divide(a: float, b: float) -> float:
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

​
测试

像测试普通函数一样测试工具： 
def test_add():
    assert add(2, 3) == 5
对于使用上下文的工具，应单独测试其核心逻辑（或使用集成式测试框架）。
Last modified on April 11, 2026