টুল চুক্তি ও এক্সিকিউশন

model --request--> [ known name? ] --> [ valid arguments? ] --> [ run, with a time limit ]
                                                                  |
model <--result--  [ envelope: ok + data, or ok=false + error ] <--+

# a recorded model request: just data. Nothing has run.
request = {"id": "c1", "name": "calculator", "arguments": {"expression": "12 * (3 + 4)"}}
print("requested:", request["name"], request["arguments"])

requested: calculator {'expression': '12 * (3 + 4)'}

export PYTHONPATH=kit

# file: kit/evidencedesk/tools.py
"""EvidenceDesk tools: registry, validation, dispatch (read-only tools)."""
from __future__ import annotations

import ast
import json
import operator
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

from pydantic import BaseModel, ConfigDict, Field, ValidationError


class ToolError(Exception):
    """An expected failure inside a tool. The code is safe to show to the model."""

    def __init__(self, code: str, message: str):
        super().__init__(message)
        self.code, self.message = code, message


class ToolArgs(BaseModel):
    """Base class for every tool's arguments: unknown fields are rejected."""

    model_config = ConfigDict(extra="forbid")


@dataclass(frozen=True)
class ToolCall:
    """The one internal shape of a model's request (a request, not an action)."""

    id: str
    name: str
    arguments: dict[str, Any]


@dataclass(frozen=True)
class ToolResult:
    """Result envelope: ok=True carries data, ok=False carries an error."""

    call_id: str
    name: str
    ok: bool
    data: Any = None
    error: dict[str, Any] | None = None

    def as_dict(self) -> dict[str, Any]:
        body = {"data": self.data} if self.ok else {"error": self.error}
        return {"call_id": self.call_id, "name": self.name, "ok": self.ok, **body}

    def as_message(self) -> dict[str, str]:
        content = json.dumps(self.as_dict(), sort_keys=True, ensure_ascii=False)
        return {"role": "tool", "tool_call_id": self.call_id, "content": content}


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

    def spec(self) -> dict[str, Any]:
        schema = self.args_model.model_json_schema()
        schema.pop("title", None)
        return {"name": self.name, "description": self.description, "parameters": schema}


def _fail(call: ToolCall, code: str, message: str, **extra: Any) -> ToolResult:
    return ToolResult(call.id, call.name[:40], False, error={"code": code, "message": message, **extra})


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        if tool.name in self._tools:
            raise ValueError(f"tool already registered: {tool.name}")
        self._tools[tool.name] = tool

    def names(self) -> list[str]:
        return sorted(self._tools)

    def specs(self) -> list[dict[str, Any]]:
        return [self._tools[n].spec() for n in self.names()]

    def execute(self, call: ToolCall, timeout: float = 2.0) -> ToolResult:
        tool = self._tools.get(call.name)
        if tool is None:
            return _fail(call, "unknown_tool", "no such tool is registered")
        try:
            args = tool.args_model.model_validate(call.arguments)
        except ValidationError as exc:
            problems = [{"loc": ".".join(str(p) for p in e["loc"]), "msg": e["msg"]} for e in exc.errors()]
            return _fail(call, "invalid_arguments", "arguments rejected", problems=problems)
        pool = ThreadPoolExecutor(max_workers=1)
        try:
            data = pool.submit(tool.handler, args).result(timeout=timeout)
        except FutureTimeout:
            return _fail(call, "timeout", f"tool did not finish within {timeout} s")
        except ToolError as exc:
            return _fail(call, exc.code, exc.message)
        except Exception:  # the exception text may hold internals: do not pass it on
            return _fail(call, "internal_error", "the tool failed")
        finally:
            pool.shutdown(wait=False)
        return ToolResult(call.id, call.name, True, data=data)


# --- calculator: read-only, deterministic, no eval() ---------------------------------------------
_BINARY = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
           ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv, ast.Mod: operator.mod}
_UNARY = {ast.UAdd: operator.pos, ast.USub: operator.neg}


def safe_eval(expression: str) -> float | int:
    """Evaluate + - * / // % on plain numbers by walking the syntax tree."""

    def walk(node: ast.AST) -> float | int:
        if isinstance(node, ast.Expression):
            return walk(node.body)
        if isinstance(node, ast.Constant) and type(node.value) in (int, float):
            return node.value
        if isinstance(node, ast.BinOp) and type(node.op) in _BINARY:
            return _BINARY[type(node.op)](walk(node.left), walk(node.right))
        if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY:
            return _UNARY[type(node.op)](walk(node.operand))
        raise ToolError("bad_expression", f"not allowed: {type(getattr(node, 'op', node)).__name__}")

    try:
        tree = ast.parse(expression, mode="eval")
        value = walk(tree)
    except SyntaxError:
        raise ToolError("bad_expression", "not a valid arithmetic expression") from None
    except ZeroDivisionError:
        raise ToolError("division_by_zero", "division by zero") from None
    return round(value, 10) if isinstance(value, float) else value


class CalculatorArgs(ToolArgs):
    expression: str = Field(min_length=1, max_length=100)


calculator_tool = Tool(
    name="calculator",
    description="Evaluate an arithmetic expression with + - * / // % and parentheses.",
    args_model=CalculatorArgs,
    handler=lambda args: {"value": safe_eval(args.expression)},
)


# --- document lookup over the corpus front matter ----------------------------------------------
class LookupArgs(ToolArgs):
    doc_id: str = Field(pattern=r"^[a-z0-9-]{3,40}$")


def read_front_matter(path: Path) -> dict[str, str]:
    lines = path.read_text(encoding="utf-8").splitlines()
    meta: dict[str, str] = {}
    for line in lines[1:]:
        if line.strip() == "---":
            break
        key, _, value = line.partition(":")
        meta[key.strip()] = value.strip()
    return meta


def make_lookup_tool(corpus_dir: Path) -> Tool:

    def handler(args: LookupArgs) -> dict[str, str]:
        for path in sorted(corpus_dir.glob("*.md")):
            meta = read_front_matter(path)
            if meta.get("id") == args.doc_id:
                return {k: meta[k] for k in ("id", "title", "version", "status", "lang")}
        raise ToolError("not_found", "no document has that id")

    return Tool("lookup_document", "Look up a help-centre document by its id.", LookupArgs, handler)


def default_registry(corpus_dir: Path) -> ToolRegistry:
    registry = ToolRegistry()
    registry.register(calculator_tool)
    registry.register(make_lookup_tool(corpus_dir))
    return registry


# --- provider shapes -> one ToolCall (dict in, dataclass out) ----------------------------------
def _arguments(raw: Any) -> dict[str, Any]:
    if isinstance(raw, str):  # some APIs send the arguments as a JSON string
        try:
            raw = json.loads(raw)
        except json.JSONDecodeError:
            raise ToolError("malformed_call", "arguments are not valid JSON") from None
    if not isinstance(raw, dict):
        raise ToolError("malformed_call", "arguments are not an object")
    return raw


def from_openai(item: dict[str, Any]) -> ToolCall:
    return ToolCall(item["call_id"], item["name"], _arguments(item["arguments"]))


def from_anthropic(block: dict[str, Any]) -> ToolCall:
    return ToolCall(block["id"], block["name"], _arguments(block["input"]))


def from_gemini(step: dict[str, Any]) -> ToolCall:
    return ToolCall(step["id"], step["name"], _arguments(step["arguments"]))


def from_ollama(call: dict[str, Any], index: int = 0) -> ToolCall:
    fn = call["function"]  # this shape has no call id: make a stable one from the position
    return ToolCall(f"call_{index}", fn["name"], _arguments(fn["arguments"]))


# --- offline: a stand-in model and one step ----------------------------------------------------
class ScriptedToolModel:
    """Fake model: replays recorded requests. Real models are less predictable."""

    def __init__(self, recorded: list[dict[str, Any]]):
        self._recorded = list(recorded)

    def next_call(self, question: str) -> dict[str, Any] | None:
        return self._recorded.pop(0) if self._recorded else None


def run_one_step(model: ScriptedToolModel, registry: ToolRegistry, question: str) -> list[dict[str, Any]]:
    transcript: list[dict[str, Any]] = [{"role": "user", "content": question}]
    raw = model.next_call(question)
    if raw is None:
        return transcript
    transcript.append({"role": "assistant", "tool_request": raw})
    arguments = raw.get("arguments")
    call = ToolCall(str(raw.get("id", "")), str(raw.get("name", "")), arguments if isinstance(arguments, dict) else {})
    transcript.append(registry.execute(call).as_message())
    return transcript

from pprint import pprint
from pathlib import Path
from evidencedesk.tools import ToolCall, default_registry

registry = default_registry(Path("kit/corpus"))
print(registry.names())
spec = registry.specs()[0]
print(spec["description"])
pprint(spec["parameters"], width=76)

['calculator', 'lookup_document']
Evaluate an arithmetic expression with + - * / // % and parentheses.
{'additionalProperties': False,
 'properties': {'expression': {'maxLength': 100,
                               'minLength': 1,
                               'title': 'Expression',
                               'type': 'string'}},
 'required': ['expression'],
 'type': 'object'}

import ast
from evidencedesk.tools import ToolError, safe_eval

tree = ast.parse("12 * (3 + 4)", mode="eval")
print([type(node).__name__ for node in ast.walk(tree)])
for text in ["12 * (3 + 4)", "7 // 2 + 0.5", "__import__('os').getcwd()", "2 ** 999999", "1 / 0", "1 +"]:
    try:
        print(f"{text!r:28} -> {safe_eval(text)}")
    except ToolError as exc:
        print(f"{text!r:28} -> refused: {exc.code} ({exc.message})")

['Expression', 'BinOp', 'Constant', 'Mult', 'BinOp', 'Constant', 'Add', 'Constant']
'12 * (3 + 4)'               -> 84
'7 // 2 + 0.5'               -> 3.5
"__import__('os').getcwd()"  -> refused: bad_expression (not allowed: Call)
'2 ** 999999'                -> refused: bad_expression (not allowed: Pow)
'1 / 0'                      -> refused: division_by_zero (division by zero)
'1 +'                        -> refused: bad_expression (not a valid arithmetic expression)

def show(name, **arguments):
    result = registry.execute(ToolCall("c1", name, arguments))
    print(name, arguments)
    print("  ->", result.error["code"] if not result.ok else result.data)
    for problem in (result.error or {}).get("problems", []):
        print("    ", problem["loc"], "|", problem["msg"])

show("calculator")
show("calculator", expression="1+1", mode="fast")
show("calculator", expression=42)
show("calculator", expression="2*21")

calculator {}
  -> invalid_arguments
     expression | Field required
calculator {'expression': '1+1', 'mode': 'fast'}
  -> invalid_arguments
     mode | Extra inputs are not permitted
calculator {'expression': 42}
  -> invalid_arguments
     expression | Input should be a valid string
calculator {'expression': '2*21'}
  -> {'value': 42}

import threading
from evidencedesk.tools import Tool, ToolArgs

show("delete_everything", path="/")
show("lookup_document", doc_id="kc-warranty-v2")
show("lookup_document", doc_id="kc-nope")
show("lookup_document", doc_id="../../etc/passwd")

release = threading.Event()   # lets the slow handler stop when we are done


class NoArgs(ToolArgs):
    pass


registry.register(Tool("slow", "waits for a long time", NoArgs, lambda args: release.wait(5)))
timed_out = registry.execute(ToolCall("c2", "slow", {}), timeout=0.05)
release.set()
print("slow ->", timed_out.error)

delete_everything {'path': '/'}
  -> unknown_tool
lookup_document {'doc_id': 'kc-warranty-v2'}
  -> {'id': 'kc-warranty-v2', 'title': 'Warranty', 'version': '2', 'status': 'current', 'lang': 'en'}
lookup_document {'doc_id': 'kc-nope'}
  -> not_found
lookup_document {'doc_id': '../../etc/passwd'}
  -> invalid_arguments
     doc_id | String should match pattern '^[a-z0-9-]{3,40}$'
slow -> {'code': 'timeout', 'message': 'tool did not finish within 0.05 s'}

ok = registry.execute(ToolCall("c3", "lookup_document", {"doc_id": "kc-returns-v2"}))
bad = registry.execute(ToolCall("c4", "calculator", {"expression": "2 ** 9"}))
import json
from pprint import pprint

for result in (ok, bad):
    pprint(json.loads(result.as_message()["content"]), width=76)

{'call_id': 'c3',
 'data': {'id': 'kc-returns-v2',
          'lang': 'en',
          'status': 'superseded',
          'title': 'Returns policy (old)',
          'version': '2'},
 'name': 'lookup_document',
 'ok': True}
{'call_id': 'c4',
 'error': {'code': 'bad_expression', 'message': 'not allowed: Pow'},
 'name': 'calculator',
 'ok': False}

from pprint import pprint
from evidencedesk.tools import ScriptedToolModel, run_one_step

model = ScriptedToolModel([{"id": "c5", "name": "lookup_document", "arguments": {"doc_id": "kc-warranty-v2"}}])
for message in run_one_step(model, registry, "Which document is the warranty?"):
    pprint(message, width=76)
print(len(run_one_step(ScriptedToolModel([]), registry, "Hello")), "message when the model asks for nothing")

{'content': 'Which document is the warranty?', 'role': 'user'}
{'role': 'assistant',
 'tool_request': {'arguments': {'doc_id': 'kc-warranty-v2'},
                  'id': 'c5',
                  'name': 'lookup_document'}}
{'content': '{"call_id": "c5", "data": {"id": "kc-warranty-v2", "lang": '
            '"en", "status": "current", "title": "Warranty", "version": '
            '"2"}, "name": "lookup_document", "ok": true}',
 'role': 'tool',
 'tool_call_id': 'c5'}
1 message when the model asks for nothing

import json
from evidencedesk.tools import from_anthropic, from_gemini, from_ollama, from_openai

# Hand-written examples that follow the documented shapes. They are NOT real responses.
args = {"expression": "12 * (3 + 4)"}
openai_item = {"type": "function_call", "call_id": "call_1", "name": "calculator", "arguments": json.dumps(args)}
anthropic_block = {"type": "tool_use", "id": "toolu_1", "name": "calculator", "input": args}
gemini_step = {"type": "function_call", "id": "fc_1", "name": "calculator", "arguments": args}
ollama_call = {"function": {"name": "calculator", "arguments": args}}

for call in (from_openai(openai_item), from_anthropic(anthropic_block), from_gemini(gemini_step), from_ollama(ollama_call)):
    print(call)

ToolCall(id='call_1', name='calculator', arguments={'expression': '12 * (3 + 4)'})
ToolCall(id='toolu_1', name='calculator', arguments={'expression': '12 * (3 + 4)'})
ToolCall(id='fc_1', name='calculator', arguments={'expression': '12 * (3 + 4)'})
ToolCall(id='call_0', name='calculator', arguments={'expression': '12 * (3 + 4)'})

# not run here: needs a provider SDK, an API key and network access
# Responses-style call (OpenAI SDK)
specs = [{"type": "function", **spec} for spec in registry.specs()]
reply = client.responses.create(model=MODEL, tools=specs, input=messages)
messages += reply.output
for item in reply.output:
    if item.type == "function_call":
        result = registry.execute(from_openai({"call_id": item.call_id, "name": item.name, "arguments": item.arguments}))
        messages.append({"type": "function_call_output", "call_id": item.call_id, "output": result.as_message()["content"]})
# the Messages-style call of the other SDK differs only in shape: from_anthropic() maps its tool_use blocks

# file: kit/tests/test_tools.py
import json
import threading
from pathlib import Path

import pytest

from evidencedesk.tools import (ScriptedToolModel, Tool, ToolArgs, ToolCall, ToolError, ToolRegistry, default_registry,
                                from_anthropic, from_gemini, from_ollama, from_openai, run_one_step, safe_eval)

CORPUS = Path(__file__).resolve().parents[1] / "corpus"


@pytest.fixture
def registry():
    return default_registry(CORPUS)


def call(name, **arguments):
    return ToolCall("c1", name, arguments)


def test_calculator_ok(registry):
    result = registry.execute(call("calculator", expression="12 * (3 + 4) - 10 / 4"))
    assert result.ok and result.data == {"value": 81.5}


@pytest.mark.parametrize("bad", ["__import__('os')", "2 ** 3", "abs(-1)", "x + 1", "'a' * 3", "True + 1", "1 +"])
def test_safe_eval_refuses_everything_but_arithmetic(bad):
    with pytest.raises(ToolError):
        safe_eval(bad)


def test_missing_extra_and_invalid_arguments(registry):
    missing = registry.execute(call("calculator"))
    extra = registry.execute(call("calculator", expression="1+1", mode="fast"))
    invalid = registry.execute(call("calculator", expression=42))
    for r in (missing, extra, invalid):
        assert r.error["code"] == "invalid_arguments"
    assert missing.error["problems"][0]["loc"] == "expression"
    assert extra.error["problems"][0]["loc"] == "mode"
    assert all("url" not in p for p in invalid.error["problems"])


def test_unknown_tool_is_rejected(registry):
    result = registry.execute(call("delete_everything", path="/"))
    assert result.error["code"] == "unknown_tool"


def test_lookup_document(registry):
    ok = registry.execute(call("lookup_document", doc_id="kc-warranty-v2"))
    assert ok.data["title"] == "Warranty" and ok.data["status"] == "current"
    assert registry.execute(call("lookup_document", doc_id="kc-nope")).error["code"] == "not_found"
    assert registry.execute(call("lookup_document", doc_id="../../etc/passwd")).error["code"] == "invalid_arguments"


def test_timeout_and_unexpected_exception():
    release = threading.Event()

    class NoArgs(ToolArgs):
        pass

    def boom(_):
        raise RuntimeError("secret internal detail")

    registry = ToolRegistry()
    registry.register(Tool("slow", "waits", NoArgs, lambda _: release.wait(5)))
    registry.register(Tool("boom", "fails", NoArgs, boom))
    try:
        assert registry.execute(call("slow"), timeout=0.05).error["code"] == "timeout"
    finally:
        release.set()
    failed = registry.execute(call("boom"))
    assert failed.error == {"code": "internal_error", "message": "the tool failed"}






def test_provider_shapes_map_to_one_call():
    args = {"expression": "1+1"}
    calls = [
        from_openai({"type": "function_call", "call_id": "a", "name": "calculator", "arguments": json.dumps(args)}),
        from_anthropic({"type": "tool_use", "id": "a", "name": "calculator", "input": args}),
        from_gemini({"type": "function_call", "id": "a", "name": "calculator", "arguments": args}),
        from_ollama({"function": {"name": "calculator", "arguments": args}}),
    ]
    assert {(c.name, json.dumps(c.arguments)) for c in calls} == {("calculator", json.dumps(args))}
    with pytest.raises(ToolError):
        from_openai({"call_id": "a", "name": "calculator", "arguments": "{not json"})


def test_safe_one_step(registry):
    model = ScriptedToolModel([{"id": "c9", "name": "lookup_document", "arguments": {"doc_id": "kc-returns-v2"}}])
    transcript = run_one_step(model, registry, "Which returns document is old?")
    body = json.loads(transcript[-1]["content"])
    assert body["ok"] and body["data"]["status"] == "superseded"
    assert len(transcript) == 3

python -m pytest -q kit/tests/test_tools.py | sed 's/ in [0-9.]*s//'

..............                                                           [100%]
14 passed

result = registry.execute(ToolCall("e1", "calculator", {"expression": "-5 + 10 % 4"}))
print(result.ok, result.data)

True {'value': -3}

from pydantic import BaseModel, ValidationError, model_validator


class Envelope(BaseModel):
    call_id: str
    name: str
    ok: bool
    data: object = None
    error: dict | None = None

    @model_validator(mode="after")
    def data_xor_error(self):
        broken = (self.data is None or self.error is not None) if self.ok else self.error is None
        if broken:
            raise ValueError("an ok result needs data and no error; a failure needs an error")
        return self


print(Envelope(**ok.as_dict()).ok, Envelope(**bad.as_dict()).ok)
try:
    Envelope(call_id="x", name="calculator", ok=True, error={"code": "x"})
except ValidationError as exc:
    print(exc.errors()[0]["msg"])

True False
Value error, an ok result needs data and no error; a failure needs an error

from pathlib import Path
from typing import Literal
from evidencedesk.tools import read_front_matter


class CountArgs(ToolArgs):
    status: Literal["current", "superseded"]


def count_documents(args):
    metas = [read_front_matter(p) for p in Path("kit/corpus").glob("*.md")]
    return {"status": args.status, "count": sum(m.get("status") == args.status for m in metas)}


registry.register(Tool("count_documents", "Count documents with a given status.", CountArgs, count_documents))
for status in ("current", "superseded", "draft"):
    result = registry.execute(ToolCall("h1", "count_documents", {"status": status}))
    print(status, "->", result.data if result.ok else result.error["problems"])

current -> {'status': 'current', 'count': 12}
superseded -> {'status': 'superseded', 'count': 1}
draft -> [{'loc': 'status', 'msg': "Input should be 'current' or 'superseded'"}]