跳转到主要内容
工具让智能体可以调用你的 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="获取用户数据(模拟 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}]
支持:基础类型、listdictLiteralEnum、可选/联合类型、Pydantic 模型、数据类、嵌套模型。 必需参数没有默认值。可选参数有默认值。

装饰器选项

from dedalus_mcp import tool

@tool(
    name="find_products",           # 覆盖工具名称
    description="Search catalog",   # 工具描述
    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="处理文件并报告进度")
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("不能被零除")
    return a / b

测试

像普通函数那样测试工具:
def test_add():
    assert add(2, 3) == 5
对于使用上下文的工具,要将核心逻辑单独测试(或使用集成式测试支架)。