toolregistry.openapi_integration module

toolregistry.openapi_integration.check_common_endpoints(url: str) → Dict[str, Any]

Check common endpoints to locate the OpenAPI schema.

This function appends a set of common endpoint paths to the provided base URL and checks if any of them return a valid response indicating the presence of an OpenAPI specification.

Parameters:

url (str) – Base URL of the web service.

Returns:

A dictionary with key “found” (bool). If a valid endpoint is found, the dictionary also contains “schema_url” (str) with the full URL of the schema.

Return type:

Dict[str, Any]

toolregistry.openapi_integration.parse_openapi_spec_from_url(url: str) → Dict[str, Any]

Retrieve and parse an OpenAPI specification from a URL.

The function first attempts to locate the schema by checking common endpoints. If a valid schema is found there, it is returned; otherwise, it falls back to the original URL.

Parameters:

url (str) – URL to the OpenAPI specification (in JSON or YAML format).

Returns:

The parsed OpenAPI specification.

Return type:

Dict[str, Any]

Raises:

ValueError – If the specification cannot be validated or parsed.

toolregistry.openapi_integration.get_openapi_spec(source: str) → Dict[str, Any]

Parse the OpenAPI specification from a file path or URL.

This function determines whether the source is a URL or a local file. For URLs, it retrieves and parses the specification over HTTP. For local files, it reads and parses the file content.

Parameters:

source (str) – The file path or URL to the OpenAPI specification (JSON/YAML).

Returns:

The fully resolved OpenAPI specification.

Return type:

Dict[str, Any]

Raises:

• FileNotFoundError – If the local file is not found.

• ValueError – If the specification cannot be parsed.

• RuntimeError – For any unexpected errors.

class toolregistry.openapi_integration.OpenAPIToolWrapper(base_url: str, name: str, method: str, path: str, params: List[str] | None)

Bases: object

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

Parameters:

• base_url (str) – The base URL of the API.

• name (str) – The name of the tool.

• method (str) – The HTTP method (e.g. “get”, “post”).

• path (str) – The API endpoint path.

• params (Optional[List[str]]) – List of parameter names for the API call.

__init__(base_url: str, name: str, method: str, path: str, params: List[str] | None) → None

call_sync(*args: Any, **kwargs: Any) → Any

Synchronously call the API using httpx.

Parameters:

• *args – Positional arguments for the API call.

• **kwargs – Keyword arguments for the API call.

Returns:

The JSON response from the API.

Return type:

Any

Raises:

• ValueError – If the base URL or tool name is not set.

• httpx.HTTPStatusError – If an HTTP error occurs.

async call_async(*args: Any, **kwargs: Any) → Any

Asynchronously call the API using httpx.

Parameters:

• *args – Positional arguments for the API call.

• **kwargs – Keyword arguments for the API call.

Returns:

The JSON response from the API.

Return type:

Any

Raises:

• ValueError – If the base URL or tool name is not set.

• httpx.HTTPStatusError – If an HTTP error occurs.

class toolregistry.openapi_integration.OpenAPITool(*, name: str, description: str, parameters: Dict[str, Any], callable: Callable[[...], Any], is_async: bool = False, parameters_model: Any | None = None)

Bases: Tool

Wrapper class for OpenAPI tools preserving function metadata.

classmethod from_openapi_spec(base_url: str, path: str, method: str, spec: Dict[str, Any], namespace: str | None = None) → OpenAPITool

Create an OpenAPITool instance from an OpenAPI specification.

Parameters:

• base_url (str) – Base URL of the service.

• path (str) – API endpoint path.

• method (str) – HTTP method.

• spec (Dict[str, Any]) – The OpenAPI operation specification.

• namespace (Optional[str]) – Optional namespace to prefix tool names with.

Returns:

An instance of OpenAPITool configured for the specified operation.

Return type:

OpenAPITool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class toolregistry.openapi_integration.OpenAPIIntegration(registry: ToolRegistry)

Bases: object

Handles integration with OpenAPI services for tool registration.

registry

The tool registry where tools are registered.

Type:

ToolRegistry

__init__(registry: ToolRegistry) → None

async register_openapi_tools_async(spec_source: str, base_url: str | None = None, with_namespace: bool | str = False) → None

Asynchronously register all tools defined in an OpenAPI specification.

Parameters:

• spec_source (str) – File path or URL to the OpenAPI specification (JSON/YAML).

• base_url (Optional[str]) – Base URL for API calls. If None, will be extracted from spec.

• with_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.

Returns:

None

register_openapi_tools(spec_source: str, base_url: str | None = None, with_namespace: bool | str = False) → None

Synchronously register all tools defined in an OpenAPI specification.

Parameters:

• spec_source (str) – File path or URL to the OpenAPI specification (JSON/YAML).

• base_url (Optional[str]) – Base URL for API calls. If None, will be extracted from spec.

• with_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.

Returns:

None