Emil Shanaty
1314314567
- conformance.py: declarative scenario runner (subset matcher, $kind/$gt/ $enum/$required assertions, call_id/session_id echo checks), reference bridge, CLI entry (python -m aicc.conformance) - bridge.py: protocol version check before parsing (protocol_mismatch), session expiry handling (session_expired), malformed tool_call -> invalid_input, emit_event() for async events - client.py: Self return type - in_process.py: drain bridge event queue after response (event delivery), Self return type - websocket.py: Self return type, best-effort shutdown - tests: 14 passing (incl. conformance scenarios, 9/9 core scenarios) - ruff: all checks pass
97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
"""Tool registration and the @tool decorator."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from aicc.protocol import ToolClass
|
|
from aicc.schema import function_schema, return_schema
|
|
|
|
ToolImpl = Callable[..., Any | Awaitable[Any]]
|
|
|
|
|
|
@dataclass
|
|
class ToolDefinition:
|
|
id: str
|
|
cls: ToolClass
|
|
description: str
|
|
fn: ToolImpl
|
|
input_schema: dict[str, Any]
|
|
output_schema: dict[str, Any]
|
|
requires_capability: str | None = None
|
|
limits: dict[str, Any] | None = None
|
|
strict_input: bool = False
|
|
|
|
|
|
def tool(
|
|
_fn: ToolImpl | None = None,
|
|
*,
|
|
id: str | None = None,
|
|
cls: ToolClass | None = None,
|
|
description: str | None = None,
|
|
requires_capability: str | None = None,
|
|
limits: dict[str, Any] | None = None,
|
|
strict_input: bool = False,
|
|
) -> Any:
|
|
"""Mark a function as an AICC tool.
|
|
|
|
Use as @tool or @tool(cls=ToolClass.SENSOR, description=...).
|
|
"""
|
|
|
|
def wrap(fn: ToolImpl) -> ToolDefinition:
|
|
tool_id = id or fn.__name__
|
|
if not tool_id.replace("_", "").isalnum() or tool_id[0].isdigit():
|
|
raise ValueError(
|
|
f"Tool id '{tool_id}' must match ^[a-z][a-z0-9_]*$ "
|
|
"(lowercase letters, digits, underscores; cannot start with digit)."
|
|
)
|
|
tool_cls = cls or _infer_class(fn)
|
|
if description:
|
|
desc = description
|
|
else:
|
|
doc = inspect.getdoc(fn) or ""
|
|
desc = doc.strip().splitlines()[0] if doc else ""
|
|
if not desc:
|
|
raise ValueError(
|
|
f"Tool '{tool_id}' requires a description (set description= or add a docstring)."
|
|
)
|
|
return ToolDefinition(
|
|
id=tool_id,
|
|
cls=tool_cls,
|
|
description=desc,
|
|
fn=fn,
|
|
input_schema=function_schema(fn),
|
|
output_schema=return_schema(fn),
|
|
requires_capability=requires_capability,
|
|
limits=limits,
|
|
strict_input=strict_input,
|
|
)
|
|
|
|
if _fn is not None and callable(_fn):
|
|
return wrap(_fn)
|
|
return wrap
|
|
|
|
|
|
def _infer_class(fn: ToolImpl) -> ToolClass:
|
|
"""Default to sensor if the tool starts with get_/read_/observe_; otherwise actuator."""
|
|
name = fn.__name__.lower()
|
|
if any(name.startswith(p) for p in ("get_", "read_", "observe_", "inspect_", "see_", "hear_", "smell_", "touch_")):
|
|
return ToolClass.SENSOR
|
|
return ToolClass.ACTUATOR
|
|
|
|
|
|
async def call_tool_impl(defn: ToolDefinition, input_data: dict[str, Any]) -> Any:
|
|
"""Invoke a registered tool's implementation with the given input."""
|
|
if defn.strict_input:
|
|
from jsonschema import validate as _validate # type: ignore
|
|
|
|
_validate(instance=input_data, schema=defn.input_schema)
|
|
result = defn.fn(**input_data) if input_data else defn.fn()
|
|
if asyncio.iscoroutine(result) or inspect.isawaitable(result):
|
|
result = await result # type: ignore[func-returns-value]
|
|
return result
|