rigging.tool

This module defines handles tool interaction with rigging generation.

ApiTool(fn: t.Callable[..., t.Any], *, _fn_to_call: t.Callable[..., t.Any] | None = None)

Source code in rigging/tool/api.py

def __init__(self, fn: t.Callable[..., t.Any], *, _fn_to_call: t.Callable[..., t.Any] | None = None) -> None:
    from rigging.prompt import Prompt

    # We support _fn_to_call for cases where the incoming function
    # for signature analysis and schema generation is different from
    # the function we actually want to call (generally internal-only)

    self.fn = fn
    self._fn_to_call = _fn_to_call or fn

    # We need to do some magic here because our Prompt object and
    # associated run function lack the context needed to construct
    # the schema at runtime - so we pass in the wrapped function for
    # attribute access and the top level Prompt.run for actual execution
    if isinstance(fn, Prompt):
        self.fn = fn.func  # type: ignore
        self._fn_to_call = fn.run

    # In the case that we are recieving a bound function which is tracking
    # an originating prompt, we can extract it from a private attribute
    elif hasattr(fn, "__rg_prompt__") and isinstance(fn.__rg_prompt__, Prompt):
        if fn.__name__ in ["run_many", "run_over"]:
            raise ValueError(
                "Only the singular Prompt.run (Prompt.bind) is supported when using prompt objects inside API tools"
            )

        self.fn = fn.__rg_prompt__.func  # type: ignore
        self._fn_to_call = fn

    self.signature = inspect.signature(self.fn)
    self.type_adapter: TypeAdapter[t.Any] = TypeAdapter(self.fn)
    _ = self.schema  # Ensure schema is valid

execute(tool_call: ToolCall) -> Message async

Executes a function call on the tool.

Source code in rigging/tool/api.py

async def execute(self, tool_call: ToolCall) -> Message:
    """Executes a function call on the tool."""

    from rigging.message import ContentTypes, Message

    if tool_call.function.name != self.name:
        raise ValueError(f"Function name {tool_call.function.name} does not match {self.name}")

    with tracer.span("Tool {name}()", name=self.name, tool_call_id=tool_call.id) as span:
        args = json.loads(tool_call.function.arguments)
        span.set_attribute("arguments", args)

        # For some reason, this will throw a coroutine unawaited warning
        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", category=RuntimeWarning)
            self.type_adapter.validate_python(args)

        if inspect.iscoroutinefunction(self._fn_to_call):
            result = await self._fn_to_call(**args)
        else:
            result = self._fn_to_call(**args)

        span.set_attribute("result", result)

    message = Message(role="tool", tool_call_id=tool_call.id)

    if isinstance(result, Message):
        message.all_content = result.all_content
    elif isinstance(result, ContentTypes):
        message.all_content = [result]
    elif isinstance(result, list) and all(isinstance(item, ContentTypes) for item in result):
        message.all_content = result
    else:
        message.all_content = str(result)

    return message

Tool

Base class for implementing internally-managed tools in the Rigging system.

You should subclass this to define your own tools:

def Hammer(Tool):
    name = "Hammer"
    description = "A tool for hitting things."

    def hit(self, target: Annotated[str, "Target of the hit") -> str:
        return f"Hit {target} with a hammer."

chat = await generator.chat(...).using(Hammer()).run()

Note

The name and description attributes are required and can be defined as class attributes or properties. If you define them as properties, you must also define a getter for them.

Note

All functions on the tool must have type hints for their parameters and use the Annotated type hint to provide a description for each parameter.

description: str instance-attribute

Description of the tool

name: str instance-attribute

Name of the tool

execute(call: ToolCall) -> ToolResult

Executes a function call on the tool.

Source code in rigging/tool/native.py

def execute(self, call: ToolCall) -> ToolResult:
    """Executes a function call on the tool."""
    try:
        content = self._execute(call)
        return ToolResult(tool=call.tool, function=call.function, error=False, content=content)
    except Exception as e:
        return ToolResult(tool=call.tool, function=call.function, error=True, content=str(e))

get_description() -> ToolDescription

Creates a full description of the tool for use in prompting

Source code in rigging/tool/native.py

def get_description(self) -> ToolDescription:
    """Creates a full description of the tool for use in prompting"""
    functions: list[ToolFunction] = []
    for method_name, method in inspect.getmembers(self.__class__, predicate=inspect.isfunction):
        if not method.__qualname__.startswith(self.__class__.__name__):
            continue

        if method_name.startswith("_"):
            continue

        signature = inspect.signature(method)

        if signature.return_annotation is inspect.Signature.empty:
            raise TypeError(f"Functions must have return type hints ({method_name})")

        if signature.return_annotation != str:
            raise TypeError(f"Functions must return strings ({method_name})")

        parameters: list[ToolParameter] = []
        for param_name, param in signature.parameters.items():
            if param_name == "self":
                continue

            formatted_name = f"{method.__name__}#{param_name}"

            if param.kind not in (
                inspect.Parameter.POSITIONAL_OR_KEYWORD,
                inspect.Parameter.KEYWORD_ONLY,
            ):
                raise TypeError(f"Parameters must be positional or keyword ({formatted_name})")

            if param.annotation is inspect.Parameter.empty:
                raise TypeError(f"Parameters must have type hints ({formatted_name})")

            if t.get_origin(param.annotation) != t.Annotated:
                raise TypeError(
                    f'Parameters must be annotated like Annotated[<type>, "<description>"] ({formatted_name})'
                )

            annotation_args = t.get_args(param.annotation)

            if len(annotation_args) != 2 or not isinstance(annotation_args[1], str):
                raise TypeError(
                    f'Parameters must be annotated like Annotated[<type>, "<description>"] ({formatted_name})'
                )

            if annotation_args[0] not in SUPPORTED_TOOL_ARGUMENT_TYPES_LIST:
                raise TypeError(
                    f"Parameters must be annotated with one of these types: {SUPPORTED_TOOL_ARGUMENT_TYPES_LIST} ({formatted_name})"
                )

            type_name = annotation_args[0].__name__
            description = annotation_args[1]

            parameters.append(ToolParameter(name=param_name, type=type_name, description=description))

        functions.append(
            ToolFunction(
                name=method_name,
                description=method.__doc__ if method.__doc__ else "",
                parameters=parameters,
            )
        )

    return ToolDescription(name=self.name, description=self.description, functions=functions)