OpenAPI 集成¶

本节介绍 ToolRegistry 库的 OpenAPI/Swagger 集成功能。

架构概览¶

OpenAPI 集成旨在基于 OpenAPI 规范自动发现和注册 REST API 端点为工具。该架构采用三层设计：

核心组件¶

OpenAPIToolWrapper：一个包装器类，提供同步和异步 HTTP 客户端方法用于 API 调用
处理 GET、POST、PUT、DELETE 请求
支持参数处理和验证
使用内置 HTTP 客户端实现 HTTP 通信
OpenAPITool：一个工具类，保留从 OpenAPI 规范中提取的函数元数据
从 OpenAPI 规范自动生成参数模式
规范化工具名称和描述
支持命名空间
OpenAPIIntegration：主集成类，协调注册流程
解析 OpenAPI 规范
为每个端点创建工具实例
支持同步和异步注册

设计模式¶

工厂模式：OpenAPITool.from_openapi_spec() 从规范创建工具实例
包装器模式：OpenAPIToolWrapper 为 HTTP 操作提供统一接口
模板方法：同步和异步版本遵循相似的模式，支持 async/await

主要特性¶

从 OpenAPI 模式自动提取参数
支持查询参数、路径参数和请求体
命名空间支持，用于工具组织
完整的 async/await 兼容性
自动 HTTP 状态错误处理

API 参考¶

OpenAPIToolWrapper¶

提供同步和异步方法用于 OpenAPI 工具调用的包装器类。

toolregistry.integrations.openapi.integration.OpenAPIToolWrapper ¶

OpenAPIToolWrapper(client_config: HttpClientConfig, name: str, method: str, path: str, params: list[str] | None, persistent: bool = True)

Bases: BaseToolWrapper

Wrapper class that provides both synchronous and asynchronous methods for OpenAPI tool calls.

Parameters:

Name	Type	Description	Default
`client_config`	`HttpClientConfig`	Configuration for the HTTP client.	required
`name`	`str`	The name of the tool.	required
`method`	`str`	The HTTP method (e.g., "get", "post").	required
`path`	`str`	The API endpoint path.	required
`params`	`Optional[List[str]]`	List of parameter names for the API call.	required
`persistent`	`bool`	If True, reuse a persistent HTTP client.	`True`

call_async `async` ¶

call_async(*args: Any, **kwargs: Any) -> Any

Asynchronously call the API using the client configuration.

Parameters:

Name	Type	Description	Default
`*args`	`Any`	Positional arguments for the API call.	`()`
`**kwargs`	`Any`	Keyword arguments for the API call.	`{}`

Returns:

Name	Type	Description
`Any`	`Any`	The JSON response from the API.

Raises:

Type	Description
`ValueError`	If the tool name is not set.
`HTTPStatusError`	If an HTTP error occurs.

call_sync ¶

call_sync(*args: Any, **kwargs: Any) -> Any

Synchronously call the API using the client configuration.

Parameters:

Name	Type	Description	Default
`*args`	`Any`	Positional arguments for the API call.	`()`
`**kwargs`	`Any`	Keyword arguments for the API call.	`{}`

Returns:

Name	Type	Description
`Any`	`Any`	The JSON response from the API.

Raises:

Type	Description
`ValueError`	If the tool name is not set.
`HTTPStatusError`	If an HTTP error occurs.

OpenAPITool¶

保留函数元数据的 OpenAPI 工具包装器类。

toolregistry.integrations.openapi.integration.OpenAPITool ¶

Bases: Tool

Wrapper class for OpenAPI tools preserving function metadata.

from_openapi_spec `classmethod` ¶

from_openapi_spec(client_config: HttpClientConfig, path: str, method: str, spec: dict[str, Any], namespace: str | None = None, persistent: bool = True) -> OpenAPITool

Create an OpenAPITool instance from an OpenAPI specification.

Parameters:

Name	Type	Description	Default
`client_config`	`HttpClientConfig`	Configuration for HTTP client.	required
`path`	`str`	API endpoint path.	required
`method`	`str`	HTTP method.	required
`spec`	`Dict[str, Any]`	The OpenAPI operation specification.	required
`namespace`	`Optional[str]`	Optional namespace to prefix tool names with.	`None`

Returns:

Name	Type	Description
`OpenAPITool`	`OpenAPITool`	An instance of OpenAPITool configured for the specified operation.

OpenAPIIntegration¶

处理与 OpenAPI 服务集成以进行工具注册的类。

toolregistry.integrations.openapi.integration.OpenAPIIntegration ¶

OpenAPIIntegration(registry: ToolRegistry)

Handles integration with OpenAPI services for tool registration.

Attributes:

Name	Type	Description
`registry`	`ToolRegistry`	The tool registry where tools are registered.

close ¶

close() -> None

Close all persistent HTTP clients (sync).

close_async `async` ¶

close_async() -> None

Close all persistent HTTP clients (async).

register_openapi_tools ¶

register_openapi_tools(client_config: HttpClientConfig, openapi_spec: dict[str, Any], namespace: bool | str = False, persistent: bool = True) -> None

Synchronously register all tools defined in an OpenAPI specification.

Parameters:

Name	Type	Description	Default
`client_config`	`HttpClientConfig`	Configuration for the HTTP client.	required
`openapi_spec`	`Dict[str, Any]`	The OpenAPI specification dictionary.	required
`namespace`	`Union[bool, str]`	Whether to prefix tool names with a namespace. - If `False`, no namespace is used. - If `True`, the namespace is derived from the OpenAPI info.title. - If a string is provided, it is used as the namespace. Defaults to False.	`False`
`persistent`	`bool`	If True (default), reuse a persistent HTTP client for connection pooling.	`True`

Returns:

Type	Description
`None`	None

register_openapi_tools_async `async` ¶

register_openapi_tools_async(client_config: HttpClientConfig, openapi_spec: dict[str, Any], namespace: bool | str = False, persistent: bool = True) -> None

Asynchronously register all tools defined in an OpenAPI specification.

Parameters:

Name	Type	Description	Default
`client_config`	`HttpClientConfig`	Configuration for the HTTP client.	required
`openapi_spec`	`Dict[str, Any]`	The OpenAPI specification dictionary.	required
`namespace`	`Union[bool, str]`	Whether to prefix tool names with a namespace. - If `False`, no namespace is used. - If `True`, the namespace is derived from the OpenAPI info.title. - If a string is provided, it is used as the namespace. Defaults to False.	`False`
`persistent`	`bool`	If True (default), reuse a persistent HTTP client for connection pooling.	`True`

Returns:

Type	Description
`None`	None

模块工具¶

OpenAPI 工具函数¶

用于 OpenAPI 处理的实用函数。

toolregistry.integrations.openapi.utils ¶

determine_urls ¶

determine_urls(url: str) -> dict[str, Any]

Determine whether the given URL or its common endpoints contain an OpenAPI schema.

Parameters:

Name	Type	Description	Default
`url`	`str`	Base URL or schema URL.	required

Returns:

Type	Description
`dict[str, Any]`	Dict[str, Any]: Contains "found" (bool), "schema_url" (str) if valid or None, and "base_api_url" (str).

extract_base_url_from_specs ¶

extract_base_url_from_specs(openapi_spec: dict[str, Any]) -> str | None

Extract and validate the base URL from the 'servers' field of the OpenAPI specification.

Parameters:

Name	Type	Description	Default
`openapi_spec`	`Dict[str, Any]`	The parsed OpenAPI specification.	required

Returns:

Type	Description
`str \| None`	Optional[str]: The validated base API URL extracted from the 'servers' field, or None if not valid.

load_openapi_spec ¶

load_openapi_spec(uri: str) -> dict[str, Any]

Sync version that calls the async implementation.

Parameters:

Name	Type	Description	Default
`uri`	`str`	URL or file path pointing to an OpenAPI specification.	required

Returns:

Type	Description
`dict[str, Any]`	Dict[str, Any]: A dictionary containing the parsed OpenAPI specification.

Raises:

Type	Description
`ValueError`	If URI retrieval, parsing, or decoding fails.

load_openapi_spec_async `async` ¶

load_openapi_spec_async(uri: str) -> dict[str, Any]

Async version of load_openapi_spec using AsyncClient.

Parameters:

Name	Type	Description	Default
`uri`	`str`	URL or file path pointing to an OpenAPI specification.	required

Returns:

Type	Description
`dict[str, Any]`	Dict[str, Any]: A dictionary containing the parsed OpenAPI specification.

Raises:

Type	Description
`ValueError`	If URI retrieval, parsing, or decoding fails.

OpenAPI 集成¶

架构概览¶

核心组件¶

设计模式¶

主要特性¶

API 参考¶

OpenAPIToolWrapper¶

toolregistry.integrations.openapi.integration.OpenAPIToolWrapper ¶

call_async async ¶

call_sync ¶

OpenAPITool¶

toolregistry.integrations.openapi.integration.OpenAPITool ¶

from_openapi_spec classmethod ¶

OpenAPIIntegration¶

toolregistry.integrations.openapi.integration.OpenAPIIntegration ¶

close ¶

close_async async ¶

register_openapi_tools ¶

register_openapi_tools_async async ¶

模块工具¶

OpenAPI 工具函数¶

toolregistry.integrations.openapi.utils ¶

determine_urls ¶

extract_base_url_from_specs ¶

load_openapi_spec ¶

load_openapi_spec_async async ¶

call_async `async` ¶

from_openapi_spec `classmethod` ¶

close_async `async` ¶

register_openapi_tools_async `async` ¶

load_openapi_spec_async `async` ¶