1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
|
"""Registry of tools exposed to the LLM."""
from __future__ import annotations
import json
from collections.abc import Callable, Sequence
from dataclasses import dataclass, field
from typing import Any
from .fetch import search_and_fetch
from .pdf import pdf_to_text
from .scrape import scrape_url
from .search import web_search
@dataclass
class ToolArtifact:
"""Binary output produced by a tool, kept outside its text result."""
kind: str
mime: str
name: str
data: bytes
file_id: str | None = None
for_user: bool = True
for_assistant: bool = True
def public_dict(self) -> dict[str, Any]:
"""Return browser-safe metadata without the binary payload."""
return {
"id": self.file_id,
"kind": self.kind,
"mime": self.mime,
"name": self.name,
"size": len(self.data),
}
@dataclass
class ToolExecutionResult:
"""A tool's LLM-facing text and separately handled binary artifacts."""
text: str
artifacts: list[ToolArtifact] = field(default_factory=list)
@dataclass
class ToolRegistry:
"""A simple registry mapping tool names to callable implementations."""
_tools: dict[str, Callable[..., object]] = field(default_factory=dict)
_ollama_tool_defs: list[dict[str, Any]] = field(default_factory=list)
_mcp_tool_names: set[str] = field(default_factory=set)
def register(self, func: Callable[..., object]) -> Callable[..., object]:
"""Register a callable under its ``__name__``."""
self._tools[func.__name__] = func
return func
def register_mcp_tool(
self,
name: str,
tool_def: dict[str, Any],
executor: Callable[..., object],
) -> Callable[..., object]:
"""Register a remote MCP tool under a normalized name.
Args:
name: The normalized ollama-visible tool name.
tool_def: The ollama-compatible tool definition dict.
executor: A callable that accepts ``(full_name, arguments)`` and
returns a JSON-encoded result string.
"""
if name in self._tools:
raise ValueError(f"Duplicate tool name: {name}")
self._tools[name] = executor
self._mcp_tool_names.add(name)
# Avoid duplicate definitions if a registry is rebuilt.
self._ollama_tool_defs = [
d for d in self._ollama_tool_defs if d.get("function", {}).get("name") != name
]
self._ollama_tool_defs.append(tool_def)
return executor
@property
def names(self) -> list[str]:
return list(self._tools.keys())
@property
def callables(self) -> list[Callable[..., object]]:
return list(self._tools.values())
@property
def ollama_tools(self) -> list[dict[str, Any] | Callable[..., object]]:
"""Return a list suitable for ollama's ``tools`` argument.
Built-in tools are passed as their callables so ollama can infer schemas
from docstrings. MCP tools are passed as pre-built dict definitions.
"""
# Only return built-in callables and dict definitions; skip the MCP
# executor callables stored in the tools dict.
return [
func for name, func in self._tools.items() if name not in self._mcp_tool_names
] + self._ollama_tool_defs
def get(self, name: str) -> Callable[..., object] | None:
return self._tools.get(name)
def execute_detailed(
self, name: str, arguments: dict[str, object] | str
) -> ToolExecutionResult:
"""Execute a tool while preserving any binary artifacts separately."""
result = self._execute_raw(name, arguments)
if isinstance(result, ToolExecutionResult):
return result
if isinstance(result, str):
return ToolExecutionResult(result)
try:
return ToolExecutionResult(json.dumps(result, ensure_ascii=False, default=str))
except (TypeError, ValueError):
return ToolExecutionResult(str(result))
def execute(self, name: str, arguments: dict[str, object] | str) -> str:
"""Execute a named tool and return a JSON-encoded string result.
Args:
name: The tool function name.
arguments: Either a dict of keyword arguments or a JSON string that
decodes to one.
Returns:
A JSON string representation of the tool's return value, or an error
object when the tool is unknown or raises.
"""
return self.execute_detailed(name, arguments).text
def _execute_raw(self, name: str, arguments: dict[str, object] | str) -> object:
"""Execute a named tool without flattening a structured result."""
func = self.get(name)
if func is None:
return json.dumps({"error": f"Unknown tool: {name}"})
if isinstance(arguments, str):
try:
arguments = json.loads(arguments) if arguments else {}
except json.JSONDecodeError:
return json.dumps({"error": f"Invalid JSON arguments for {name}: {arguments}"})
try:
if name in self._mcp_tool_names:
# MCP executor expects (full_name, arguments).
result = func(name, arguments)
else:
result = func(**arguments)
except Exception as exc: # noqa: BLE001
return json.dumps({"error": f"{type(exc).__name__}: {exc}"})
return result
def default_registry() -> ToolRegistry:
"""Build a registry pre-loaded with the built-in tools."""
reg = ToolRegistry()
reg.register(web_search)
reg.register(scrape_url)
reg.register(search_and_fetch)
reg.register(pdf_to_text)
return reg
def all_tool_callables(registry: ToolRegistry | None = None) -> Sequence[Callable[..., object]]:
"""Return the list of callables suitable for ``ollama`` ``tools`` argument."""
reg = registry or default_registry()
return reg.callables
|
