API reference¶
Generated from the apogee-ai-tools source with mkdocstrings. Every symbol below is exported from apogee_ai_tools, so it is part of the supported public surface.
Application · DTOs¶
dataclass
¶
dataclass
¶
Python
CallDTO(name: str, arguments: dict[str, Any] = dict(), timeout_s: float | None = None, max_attempts: int = 1)
class-attribute
instance-attribute
¶
dataclass
¶
Application · Use cases¶
Synthetic load: call two tools alternately and report rate.
async
¶
Source code in apogee_ai_tools/application/use_cases/bench_calls_use_case.py
Python
async def execute(self, iterations: int) -> dict[str, float]:
if iterations <= 0:
raise ValueError("iterations must be positive")
registry = ToolRegistry([_bench_add, _bench_echo])
executor = AsyncExecutor(registry)
ok = 0
start = time.perf_counter()
for i in range(iterations):
if i % 2 == 0:
result = await executor.execute("add", {"a": i, "b": 1})
else:
result = await executor.execute("echo", {"message": f"hi-{i}"})
if result.succeeded:
ok += 1
elapsed = (time.perf_counter() - start) * 1000.0
return {
"iterations": float(iterations),
"successful": float(ok),
"elapsed_ms": elapsed,
"calls_per_second": (iterations / elapsed * 1000.0) if elapsed > 0 else 0.0,
"success_rate": ok / iterations,
}
Source code in apogee_ai_tools/application/use_cases/call_tool_use_case.py
async
¶
Python
execute(name: str, arguments: dict[str, Any], timeout_s: float | None = None, max_attempts: int = 1) -> ToolResult
Source code in apogee_ai_tools/application/use_cases/call_tool_use_case.py
Python
async def execute(
self,
name: str,
arguments: dict[str, Any],
timeout_s: float | None = None,
max_attempts: int = 1,
) -> ToolResult:
executor = AsyncExecutor(
self._registry, validator=self._validator, timeout_s=timeout_s,
)
if max_attempts > 1:
executor = RetryExecutor(executor, max_attempts=max_attempts)
return await executor.execute(name, arguments)
Source code in apogee_ai_tools/application/use_cases/describe_tool_use_case.py
async
¶
Source code in apogee_ai_tools/application/use_cases/describe_tool_use_case.py
Source code in apogee_ai_tools/application/use_cases/list_tools_use_case.py
async
¶
Source code in apogee_ai_tools/application/use_cases/translate_tools_use_case.py
async
¶
Source code in apogee_ai_tools/application/use_cases/translate_tools_use_case.py
Python
async def execute(self, provider: str, tag: str | None = None) -> list[dict]:
tools = self._registry.find_by_tag(tag) if tag else self._registry.all()
if provider == "openai":
return OpenAIToolsTranslator().translate(tools)
if provider == "anthropic":
return AnthropicToolsTranslator().translate(tools)
raise ToolError(f"unknown provider: {provider!r}")
Domain¶
dataclass
¶
dataclass
¶
Python
ToolDefinition(name: str, description: str, parameters: tuple[ToolParameter, ...] = (), func: Callable[..., Any] | None = None, tags: tuple[str, ...] = (), metadata: dict[str, str] = dict())
class-attribute
instance-attribute
¶
Source code in apogee_ai_tools/domain/entities/tool_definition.py
Python
def to_json_schema(self) -> dict[str, Any]:
properties = {p.name: p.to_json_schema() for p in self.parameters}
required = [p.name for p in self.parameters if p.required]
return {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": properties,
"required": required,
},
}
dataclass
¶
Python
ToolParameter(name: str, kind: ParameterKind, description: str = '', required: bool = True, default: Any = None, enum: tuple[Any, ...] = (), items_kind: ParameterKind | None = None, properties: dict[str, 'ToolParameter'] = dict())
class-attribute
instance-attribute
¶
Source code in apogee_ai_tools/domain/value_objects/tool_parameter.py
Python
def to_json_schema(self) -> dict[str, Any]:
schema: dict[str, Any] = {"type": self.kind.value}
if self.description:
schema["description"] = self.description
if self.enum:
schema["enum"] = list(self.enum)
if self.kind is ParameterKind.ARRAY and self.items_kind is not None:
schema["items"] = {"type": self.items_kind.value}
if self.kind is ParameterKind.OBJECT and self.properties:
schema["properties"] = {
k: v.to_json_schema() for k, v in self.properties.items()
}
schema["required"] = [k for k, v in self.properties.items() if v.required]
if self.default is not None and not self.required:
schema["default"] = self.default
return schema
dataclass
¶
Python
ToolResult(name: str, status: ToolCallStatus = SUCCESS, value: Any = None, error_message: str = '', latency_ms: float = 0.0, attempts: int = 1, call_id: str | None = None, metadata: dict[str, str] = dict())
class-attribute
instance-attribute
¶
Domain · Enums¶
Bases: str, Enum
Domain · Exceptions¶
Bases: Exception
Base for apogee-ai-tools errors.
Bases: ToolError
Source code in apogee_ai_tools/domain/exceptions/tool_exceptions.py
Bases: ToolError
Source code in apogee_ai_tools/domain/exceptions/tool_exceptions.py
Bases: ToolError
Source code in apogee_ai_tools/domain/exceptions/tool_exceptions.py
Domain · Protocols (ports)¶
Bases: Protocol
async
¶
Python
execute(tool_name: str, arguments: dict[str, Any]) -> ToolResult
Bases: Protocol
Python
validate(tool: ToolDefinition, arguments: dict[str, Any]) -> None
Bases: Protocol
Python
register(tool: ToolDefinition) -> None
Python
get(name: str) -> ToolDefinition
Python
all() -> Iterable[ToolDefinition]
Python
find_by_tag(tag: str) -> list[ToolDefinition]
Infrastructure¶
Outputs tools in Anthropic Messages API tools format.
Python
translate(tools: Iterable[ToolDefinition]) -> list[dict]
Source code in apogee_ai_tools/infrastructure/translators/anthropic_translator.py
Source code in apogee_ai_tools/infrastructure/executors/async_executor.py
async
¶
Python
execute(tool_name: str, arguments: dict[str, Any]) -> ToolResult
Source code in apogee_ai_tools/infrastructure/executors/async_executor.py
Python
async def execute(
self, tool_name: str, arguments: dict[str, Any]
) -> ToolResult:
tool = self._registry.get(tool_name)
try:
self._validator.validate(tool, arguments)
except ToolValidationException as exc:
return ToolResult(
name=tool_name,
status=ToolCallStatus.VALIDATION_FAILED,
error_message=str(exc),
)
if tool.func is None:
return ToolResult(
name=tool_name,
status=ToolCallStatus.FAILED,
error_message="tool has no callable",
)
start = time.perf_counter()
try:
if inspect.iscoroutinefunction(tool.func):
coro = tool.func(**arguments)
if self._timeout is not None:
value = await asyncio.wait_for(coro, timeout=self._timeout)
else:
value = await coro
else:
value = tool.func(**arguments)
except asyncio.TimeoutError:
return ToolResult(
name=tool_name,
status=ToolCallStatus.TIMEOUT,
error_message=str(ToolTimeoutException(tool_name, self._timeout or 0.0)),
latency_ms=(time.perf_counter() - start) * 1000.0,
)
except Exception as exc: # noqa: BLE001
return ToolResult(
name=tool_name,
status=ToolCallStatus.FAILED,
error_message=str(ToolExecutionException(tool_name, exc)),
latency_ms=(time.perf_counter() - start) * 1000.0,
)
return ToolResult(
name=tool_name,
status=ToolCallStatus.SUCCESS,
value=value,
latency_ms=(time.perf_counter() - start) * 1000.0,
)
Lightweight validator for ToolDefinition.parameters.
Python
validate(tool: ToolDefinition, arguments: dict[str, Any]) -> None
Source code in apogee_ai_tools/infrastructure/schema/builtin_validator.py
Python
def validate(
self, tool: ToolDefinition, arguments: dict[str, Any]
) -> None:
param_index: dict[str, ToolParameter] = {p.name: p for p in tool.parameters}
for p in tool.parameters:
if p.required and p.name not in arguments:
raise ToolValidationException(
tool.name, f"missing required parameter: {p.name!r}"
)
for name, value in arguments.items():
if name not in param_index:
raise ToolValidationException(
tool.name, f"unknown parameter: {name!r}"
)
p = param_index[name]
if not _matches(value, p.kind):
raise ToolValidationException(
tool.name,
f"{name!r} expected {p.kind.value}, got {type(value).__name__}",
)
if p.enum and value not in p.enum:
raise ToolValidationException(
tool.name,
f"{name!r}={value!r} not in enum {list(p.enum)}",
)
if p.kind is ParameterKind.ARRAY and p.items_kind is not None:
for i, item in enumerate(value):
if not _matches(item, p.items_kind):
raise ToolValidationException(
tool.name,
f"{name}[{i}] expected {p.items_kind.value}",
)
Lazy adapter for jsonschema. Install via extras=jsonschema.
Source code in apogee_ai_tools/infrastructure/schema/jsonschema_validator.py
Python
validate(tool: ToolDefinition, arguments: dict[str, Any]) -> None
Source code in apogee_ai_tools/infrastructure/schema/jsonschema_validator.py
Python
def validate(
self, tool: ToolDefinition, arguments: dict[str, Any]
) -> None:
self._ensure()
schema = tool.to_json_schema()["parameters"]
validator = self._validator_cls(schema) # type: ignore[misc]
errors = list(validator.iter_errors(arguments))
if errors:
raise ToolValidationException(
tool.name, "; ".join(e.message for e in errors)
)
Outputs tools in OpenAI Chat Completions tools format.
Python
translate(tools: Iterable[ToolDefinition]) -> list[dict]
Source code in apogee_ai_tools/infrastructure/translators/openai_translator.py
Wraps another executor with N retries on FAILED/TIMEOUT.
VALIDATION_FAILED is never retried — bad arguments will keep failing.
Source code in apogee_ai_tools/infrastructure/executors/retry_executor.py
async
¶
Python
execute(tool_name: str, arguments: dict[str, Any]) -> ToolResult
Source code in apogee_ai_tools/infrastructure/executors/retry_executor.py
Python
async def execute(
self, tool_name: str, arguments: dict[str, Any]
) -> ToolResult:
last: ToolResult | None = None
for attempt in range(1, self._max_attempts + 1):
result = await self._inner.execute(tool_name, arguments)
last = result
if result.status is ToolCallStatus.SUCCESS:
return replace(result, attempts=attempt)
if result.status is ToolCallStatus.VALIDATION_FAILED:
return replace(result, attempts=attempt)
assert last is not None
return replace(last, attempts=self._max_attempts)
Source code in apogee_ai_tools/infrastructure/executors/sync_executor.py
Python
execute(tool_name: str, arguments: dict[str, Any]) -> ToolResult
Python
ToolRegistry(tools: Iterable[ToolDefinition] | None = None)
Source code in apogee_ai_tools/infrastructure/registries/tool_registry.py
Python
register(tool: ToolDefinition) -> None
Python
get(name: str) -> ToolDefinition
Python
all() -> list[ToolDefinition]
Python
find_by_tag(tag: str) -> list[ToolDefinition]
Python
function_to_tool(func: Callable[..., Any], name: str | None = None, description: str | None = None, tags: tuple[str, ...] = ()) -> ToolDefinition
Inspects a function's type hints + docstring to build a ToolDefinition.
Source code in apogee_ai_tools/infrastructure/tools/_introspection.py
Python
def function_to_tool(
func: Callable[..., Any],
name: str | None = None,
description: str | None = None,
tags: tuple[str, ...] = (),
) -> ToolDefinition:
"""Inspects a function's type hints + docstring to build a ToolDefinition."""
signature = inspect.signature(func)
summary, doc_params = _parse_docstring(inspect.getdoc(func))
final_description = (description or summary or "").strip()
# Resolve forward references / PEP-563 string annotations.
try:
hints = typing.get_type_hints(func)
except Exception: # noqa: BLE001
hints = {}
parameters: list[ToolParameter] = []
for pname, param in signature.parameters.items():
if pname in {"self", "cls"}:
continue
annotation = hints.get(pname, param.annotation)
kind = _kind_from_annotation(annotation)
items = _items_kind(annotation) if kind is ParameterKind.ARRAY else None
parameters.append(
ToolParameter(
name=pname,
kind=kind,
description=doc_params.get(pname, ""),
required=param.default is inspect._empty, # noqa: SLF001
default=None if param.default is inspect._empty else param.default, # noqa: SLF001
items_kind=items,
)
)
return ToolDefinition(
name=name or func.__name__,
description=final_description,
parameters=tuple(parameters),
func=func,
tags=tags,
)
Python
tool(name: str | None = None, description: str | None = None, tags: tuple[str, ...] = ()) -> Callable[[Callable[..., Any]], ToolDefinition]
Decorator that turns a Python function into a ToolDefinition.
The function is preserved as definition.func so executors can call
it. The decorator returns the ToolDefinition (not the function).
Source code in apogee_ai_tools/infrastructure/tools/decorator.py
Python
def tool(
name: str | None = None,
description: str | None = None,
tags: tuple[str, ...] = (),
) -> Callable[[Callable[..., Any]], ToolDefinition]:
"""Decorator that turns a Python function into a ToolDefinition.
The function is preserved as ``definition.func`` so executors can call
it. The decorator returns the ToolDefinition (not the function).
"""
def decorate(func: Callable[..., Any]) -> ToolDefinition:
return function_to_tool(func, name=name, description=description, tags=tags)
return decorate