Helper Classes¶

Utility classes and helper functions that support the core functionality of the ToolRegistry library.

Parameter Models¶

Parameter validation and schema generation for tool functions. ArgModelBase dynamically creates Pydantic models from function signatures for runtime argument validation.

toolregistry.parameter_models ¶

ArgModelBase ¶

Bases: BaseModel

Base model for function argument validation with Pydantic.

Features

Supports arbitrary types in fields
Provides method to dump fields one level deep
Configures Pydantic model behavior

model_dump_one_level ¶

model_dump_one_level() -> dict[str, Any]

Dump model fields one level deep, keeping sub-models as-is.

Returns:

Type	Description
`dict[str, Any]`	Dict[str, Any]: Dictionary of field names to values.

Source code in src/toolregistry/parameter_models.py

def model_dump_one_level(self) -> dict[str, Any]:
    """Dump model fields one level deep, keeping sub-models as-is.

    Returns:
        Dict[str, Any]: Dictionary of field names to values.
    """
    return {field: getattr(self, field) for field in self.__pydantic_fields__}

InvalidSignature ¶

Bases: Exception

Exception raised when a function signature cannot be processed for FastMCP.

Attributes:

Name	Type	Description
`message`	`str`	Explanation of the error.

Utilities¶

Common utility functions used across the library, including tool name normalization and HTTP client configuration for OpenAPI integrations.

toolregistry.utils ¶

HttpClientConfig ¶

HttpClientConfig(base_url: str, headers: dict[str, str] | None = None, timeout: float = 10.0, auth: tuple[str, str] | None = None, **extra_options: Any)

Container for HTTP client configuration.

Creates zero-dependency HTTP clients with base URL support, connection pooling, and both sync/async interfaces.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	The base URL for the API. This is required.	required
`headers`	`dict[str, str] \| None`	Custom request headers. Default is None.	`None`
`timeout`	`float`	Request timeout in seconds. Default is 10.0.	`10.0`
`auth`	`tuple[str, str] \| None`	Basic authentication credentials (username, password). Default is None.	`None`
`**extra_options`	`Any`	Additional client parameters (e.g. verify, proxy, pool_size).	`{}`

Source code in src/toolregistry/utils.py

def __init__(
    self,
    base_url: str,
    headers: dict[str, str] | None = None,
    timeout: float = 10.0,
    auth: tuple[str, str] | None = None,
    **extra_options: Any,
):
    self.base_url = base_url.rstrip("/")
    self.headers = headers or {}
    self.timeout = timeout
    self.auth = auth
    self.extra_options = extra_options
    self._sync_client: _BaseUrlClient | None = None
    self._async_client: _BaseUrlAsyncClient | None = None

getstate ¶

__getstate__() -> dict

Drop live HTTP clients before pickling.

The config (base_url, headers, timeout, auth, extra_options) is preserved. Live _sync_client / _async_client (holding thread locks and connection pools) are dropped. The worker process will recreate them lazily via get_persistent_client().

Source code in src/toolregistry/utils.py

def __getstate__(self) -> dict:
    """Drop live HTTP clients before pickling.

    The config (base_url, headers, timeout, auth, extra_options)
    is preserved.  Live ``_sync_client`` / ``_async_client``
    (holding thread locks and connection pools) are dropped.
    The worker process will recreate them lazily via
    ``get_persistent_client()``.
    """
    return {
        "base_url": self.base_url,
        "headers": self.headers,
        "timeout": self.timeout,
        "auth": self.auth,
        "extra_options": self.extra_options,
    }

setstate ¶

__setstate__(state: dict) -> None

Restore config; clients will be created lazily on first use.

Source code in src/toolregistry/utils.py

def __setstate__(self, state: dict) -> None:
    """Restore config; clients will be created lazily on first use."""
    self.base_url = state["base_url"]
    self.headers = state["headers"]
    self.timeout = state["timeout"]
    self.auth = state["auth"]
    self.extra_options = state["extra_options"]
    self._sync_client = None
    self._async_client = None

close ¶

close()

Close persistent clients (sync).

Only closes the sync client. Use close_async() to properly close the async client.

Source code in src/toolregistry/utils.py

def close(self):
    """Close persistent clients (sync).

    Only closes the sync client. Use ``close_async()`` to properly
    close the async client.
    """
    if self._sync_client:
        self._sync_client.close()
        self._sync_client = None

close_async `async` ¶

close_async()

Close all persistent clients.

Source code in src/toolregistry/utils.py

async def close_async(self):
    """Close all persistent clients."""
    if self._async_client:
        await self._async_client.aclose()
        self._async_client = None
    if self._sync_client:
        self._sync_client.close()
        self._sync_client = None

get_persistent_client ¶

get_persistent_client(use_async: Literal[False] = False) -> _BaseUrlClient

get_persistent_client(use_async: Literal[True] = ...) -> _BaseUrlAsyncClient

get_persistent_client(use_async: bool = False)

Get or create a persistent client instance.

Unlike to_client(), this method reuses the same client across multiple calls, enabling HTTP connection pooling.

Parameters:

Name	Type	Description	Default
`use_async`	`bool`	Whether to return an async client.	`False`

Returns:

Type	Description
	A persistent _BaseUrlClient or _BaseUrlAsyncClient instance.

Source code in src/toolregistry/utils.py

def get_persistent_client(self, use_async: bool = False):
    """Get or create a persistent client instance.

    Unlike ``to_client()``, this method reuses the same client across
    multiple calls, enabling HTTP connection pooling.

    Args:
        use_async: Whether to return an async client.

    Returns:
        A persistent _BaseUrlClient or _BaseUrlAsyncClient instance.
    """
    if use_async:
        if self._async_client is None:
            self._async_client = self._make_client(use_async=True)
        return self._async_client
    else:
        if self._sync_client is None:
            self._sync_client = self._make_client(use_async=False)
        return self._sync_client

to_client ¶

to_client(use_async: Literal[False]) -> _BaseUrlClient

to_client(use_async: Literal[True]) -> _BaseUrlAsyncClient

to_client(use_async: bool = False)

Create a new HTTP client instance.

Parameters:

Name	Type	Description	Default
`use_async`	`bool`	Whether to create an asynchronous client. Default is False.	`False`

Returns:

Type	Description
	A _BaseUrlClient or _BaseUrlAsyncClient wrapping the underlying
	zero-dependency client with base URL support.

Source code in src/toolregistry/utils.py

def to_client(self, use_async: bool = False):
    """Create a new HTTP client instance.

    Args:
        use_async: Whether to create an asynchronous client. Default is False.

    Returns:
        A _BaseUrlClient or _BaseUrlAsyncClient wrapping the underlying
        zero-dependency client with base URL support.
    """
    return self._make_client(use_async=use_async)

HttpxClientConfig ¶

HttpxClientConfig(*args: Any, **kwargs: Any) -> HttpClientConfig

Deprecated alias for HttpClientConfig.

.. deprecated:: Use HttpClientConfig instead. HttpxClientConfig will be removed in a future release.

Source code in src/toolregistry/utils.py

def HttpxClientConfig(*args: Any, **kwargs: Any) -> HttpClientConfig:
    """Deprecated alias for HttpClientConfig.

    .. deprecated::
        Use ``HttpClientConfig`` instead. ``HttpxClientConfig`` will be
        removed in a future release.
    """
    warnings.warn(
        "HttpxClientConfig is deprecated, use HttpClientConfig instead",
        DeprecationWarning,
        stacklevel=2,
    )
    return HttpClientConfig(*args, **kwargs)

generate_invocation_id ¶

generate_invocation_id(prefix: Literal['bat', 'ptc', 'sig'] = 'sig') -> str

Generate a unique invocation ID for grouping related tool calls.

Prefixes

tr_bat_ — batch execution via execute_tool_calls()
tr_ptc_ — PTC code execution via PtcTool
tr_sig_ — single invocation via registry.invoke()

Parameters:

Name	Type	Description	Default
`prefix`	`Literal['bat', 'ptc', 'sig']`	One of `"bat"`, `"ptc"`, `"sig"`.	`'sig'`

Returns:

Type	Description
`str`	A unique ID like `"tr_bat_a1b2c3d4"`.

Source code in src/toolregistry/utils.py

def generate_invocation_id(prefix: Literal["bat", "ptc", "sig"] = "sig") -> str:
    """Generate a unique invocation ID for grouping related tool calls.

    Prefixes:
        - ``tr_bat_`` — batch execution via ``execute_tool_calls()``
        - ``tr_ptc_`` — PTC code execution via ``PtcTool``
        - ``tr_sig_`` — single invocation via ``registry.invoke()``

    Args:
        prefix: One of ``"bat"``, ``"ptc"``, ``"sig"``.

    Returns:
        A unique ID like ``"tr_bat_a1b2c3d4"``.
    """
    return f"tr_{prefix}_{uuid.uuid4().hex[:8]}"

normalize_tool_name ¶

normalize_tool_name(name: str) -> str

Normalize tool name to snake_case format and remove dots and spaces. Also handles OpenAPI-style duplicate names like 'add_add_get' by converting to 'add_get'.

Parameters:

Name	Type	Description	Default
`name`	`str`	Original tool name in various formats (including CamelCase, UpperCamelCase, or containing spaces)	required

Returns:

Name	Type	Description
`str`	`str`	Normalized name in snake_case without dots or spaces

Source code in src/toolregistry/utils.py

def normalize_tool_name(name: str) -> str:
    """Normalize tool name to snake_case format and remove dots and spaces.
    Also handles OpenAPI-style duplicate names like 'add_add_get' by converting to 'add_get'.

    Args:
        name: Original tool name in various formats (including CamelCase, UpperCamelCase, or containing spaces)

    Returns:
        str: Normalized name in snake_case without dots or spaces
    """
    # First check for OpenAPI-style duplicate names (e.g. "add_add_get")
    openapi_pattern = r"^([a-zA-Z0-9]+)_\1_([a-zA-Z0-9]+)$"
    match = re.match(openapi_pattern, name)
    if match:
        return f"{match.group(1)}_{match.group(2)}"

    # Replace all special chars (., -, @, etc.) with single underscore
    name = re.sub(r"[.\-@]+", "_", name)

    # Remove spaces and collapse multiple spaces into a single space
    name = re.sub(r"\s+", " ", name).strip()

    # Replace spaces with underscores
    name = name.replace(" ", "_")

    # Convert CamelCase and UpperCamelCase to snake_case
    # Handles all cases including:
    # XMLParser -> xml_parser
    # getUserIDFromDB -> get_user_id_from_db
    # HTTPRequest -> http_request
    name = re.sub(r"(?<!^)(?=[A-Z][a-z])|(?<=[a-z0-9])(?=[A-Z])", "_", name).lower()

    # Collapse multiple underscores into single underscore
    return re.sub(r"_+", "_", name)

Helper Classes¶

Parameter Models¶

toolregistry.parameter_models ¶

ArgModelBase ¶

model_dump_one_level ¶

InvalidSignature ¶

Utilities¶

toolregistry.utils ¶

HttpClientConfig ¶

__getstate__ ¶

__setstate__ ¶

close ¶

close_async async ¶

get_persistent_client ¶

to_client ¶

HttpxClientConfig ¶

generate_invocation_id ¶

normalize_tool_name ¶

getstate ¶

setstate ¶

close_async `async` ¶