跳转至

辅助类

支持 ToolRegistry 库核心功能的实用类和辅助函数。

参数模型

工具函数的参数验证和 Schema 生成。ArgModelBase 根据函数签名动态创建 Pydantic 模型,用于运行时参数验证。

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.

实用工具

库中通用的实用函数,包括工具名称标准化和用于 OpenAPI 集成的 HTTP 客户端配置。

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

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)

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)