src/agentcade/tools.py

import asyncio
import json
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, create_model

import observability


class ToolInput(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)


class NoInput(ToolInput):
    pass


class TextInput(ToolInput):
    text: str = Field(min_length=1, description="用户当前消息的完整原文")


def original_text_input(text: str):
    return create_model(
        "CurrentMessageInput", __base__=TextInput,
        text=(Literal[text], Field(description="本轮用户消息的完整原文，保持字词和标点")),
    )


@dataclass(frozen=True)
class Tool:
    name: str
    description: str
    inputs: type[BaseModel]
    handler: Callable[..., Awaitable[Any]]

    def definition(self):
        return {
            "name": self.name, "description": self.description,
            "parameters": self.inputs.model_json_schema(),
        }


class ToolRegistry:
    def __init__(self, tools: list[Tool], *, terminal=()):
        self.tools = {tool.name: tool for tool in tools}
        if len(self.tools) != len(tools):
            raise ValueError("工具名称不能重复")
        self.calls: list[dict] = []
        self.lock = asyncio.Lock()
        self.terminal = set(terminal)
        self.completed = asyncio.Event()

    def definitions(self):
        return [tool.definition() for tool in self.tools.values()]

    async def call(self, name: str, arguments: dict):
        async with self.lock:
            if self.completed.is_set():
                raise ValueError("当前工具调用轮次已经结束")
            with observability.tool_call(name, arguments) as outcome:
                tool = self.tools[name]
                inputs = tool.inputs.model_validate(arguments)
                result = await tool.handler(**inputs.model_dump())
                # 在工具执行位置验证返回值可以通过模型接口传输。
                json.dumps(result, ensure_ascii=False, allow_nan=False)
                outcome["result"] = result
            self.calls.append({"name": name, "arguments": arguments, "result": result})
            if name in self.terminal:
                self.completed.set()
            return result