Skip to content

orchestr8.adapter

BaseAdapter

Base class to be inherited by all adapters.

Source code in orchestr8/adapter/base.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class BaseAdapter:
    """Base class to be inherited by all adapters."""

    def __new__(cls, *args: Any, **kwargs: Any) -> BaseAdapter:
        if cls is BaseAdapter:
            raise TypeError("BaseAdapter cannot be instantiated directly")
        return super().__new__(cls)

    @property
    def name(self) -> str:
        """Name of the adapted object."""
        raise NotImplementedError("Parent class must implemented this property")

    @property
    def description(self) -> str | None:
        """Description of the adapted object."""
        raise NotImplementedError("Parent class must implemented this property")

    @property
    def schema(self) -> Dict[str, Any]:
        """Get the JSON schema for the adapted object."""
        raise NotImplementedError("Parent class must implemented this property")

    @property
    def openai_schema(self) -> Dict[str, Any]:
        """Schema of the function in OpenAI function calling format."""
        return to_openai_function_calling_format(self.name, self.schema)

    @property
    def anthropic_schema(self) -> Dict[str, Any]:
        """Schema of the function in Anthropic function calling format."""
        return to_anthropic_function_calling_format(self.name, self.schema)

    @property
    def gemini_schema(self) -> Dict[str, Any]:
        """Schema of the function in Gemini function calling format."""
        return to_gemini_function_calling_format(self.name, self.schema)

    def validate_input(self, input: str | Dict[str, Any], **kwargs: Any) -> Any:  # noqa: A002
        """
        Validate the input against the schema of the function.

        Args:
            input: The JSON value or dictionary object to validate.
            kwargs: Additional keyword arguments.

        Returns:
            The validated input.
        """
        raise NotImplementedError("Parent class must implemented this method")

anthropic_schema: Dict[str, Any] property

Schema of the function in Anthropic function calling format.

description: str | None property

Description of the adapted object.

gemini_schema: Dict[str, Any] property

Schema of the function in Gemini function calling format.

name: str property

Name of the adapted object.

openai_schema: Dict[str, Any] property

Schema of the function in OpenAI function calling format.

schema: Dict[str, Any] property

Get the JSON schema for the adapted object.

validate_input

Validate the input against the schema of the function.

Parameters:

Name Type Description Default
input str | Dict[str, Any]

The JSON value or dictionary object to validate.

 required
kwargs Any

Additional keyword arguments.

 {}

Returns:

Type Description
Any

The validated input.
Source code in orchestr8/adapter/base.py 
52
53
54
55
56
57
58
59
60
61
62
63
def validate_input(self, input: str | Dict[str, Any], **kwargs: Any) -> Any:  # noqa: A002
    """
    Validate the input against the schema of the function.

    Args:
        input: The JSON value or dictionary object to validate.
        kwargs: Additional keyword arguments.

    Returns:
        The validated input.
    """
    raise NotImplementedError("Parent class must implemented this method")

StructAdapter

A wrapper around pydantic's TypeAdapter for schema and definition generation with capability to validate input from python or JsON object.

Structured types such as Pydantic Models, TypedDict, dataclasses, etc. are supported.

Provides: - Code definition extraction. - Dynamic schema generation in standard and function-calling formats such as OpenAI, Anthropic, and Gemini. - Extraction of docstring metadata for populating schema. - Automatic validation of python or JsON input against the schema.

Useful for schema-driven representations of complex types for use in function calling, serialization, and documentation contexts.

from typing_extensions import TypedDict
from orchestract.adapter.struct import StructAdapter

class User(TypedDict): # Can be pydantic model or dataclass
    id: str
    role: Literal["admin", "user"]

# Create adapter
struct = StructAdapter(User)

# Normal invocation
struct(id="user123", role="user") # returns: {"id": "user123", "role": "user"}

# Access adapter properties
print(f"Struct name: {struct.name}")
print(f"Struct description: {struct.description}")
print(f"Struct docstring: {struct.docstring}")
print(f"Struct definition: {struct.definition}")
print(f"Struct schema: {struct.schema}")
print(f"Struct OpenAI schema: {struct.openai_schema}")
print(f"Struct Anthropic schema: {struct.anthropic_schema}")
print(f"Struct Gemini schema: {struct.gemini_schema}")

# Validate python input
struct.validate_input({"id": "user123", "role": "user"})

# Validate JSON input
struct.validate_input('{"id": "user123", "role": "user"}')
Source code in orchestr8/adapter/struct.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class StructAdapter(Generic[T], BaseAdapter):
    """
    A wrapper around pydantic's `TypeAdapter` for schema and definition generation with capability to
    validate input from python or JsON object.

    Structured types such as Pydantic Models, TypedDict, dataclasses, etc. are supported.

    Provides:
    - Code definition extraction.
    - Dynamic schema generation in standard and function-calling formats such as OpenAI, Anthropic, and Gemini.
    - Extraction of docstring metadata for populating schema.
    - Automatic validation of python or JsON input against the schema.

    Useful for schema-driven representations of complex types for use in function calling,
    serialization, and documentation contexts.

    ```python
    from typing_extensions import TypedDict
    from orchestract.adapter.struct import StructAdapter

    class User(TypedDict): # Can be pydantic model or dataclass
        id: str
        role: Literal["admin", "user"]

    # Create adapter
    struct = StructAdapter(User)

    # Normal invocation
    struct(id="user123", role="user") # returns: {"id": "user123", "role": "user"}

    # Access adapter properties
    print(f"Struct name: {struct.name}")
    print(f"Struct description: {struct.description}")
    print(f"Struct docstring: {struct.docstring}")
    print(f"Struct definition: {struct.definition}")
    print(f"Struct schema: {struct.schema}")
    print(f"Struct OpenAI schema: {struct.openai_schema}")
    print(f"Struct Anthropic schema: {struct.anthropic_schema}")
    print(f"Struct Gemini schema: {struct.gemini_schema}")

    # Validate python input
    struct.validate_input({"id": "user123", "role": "user"})

    # Validate JSON input
    struct.validate_input('{"id": "user123", "role": "user"}')
    ```
    """

    ref_type: Literal["function", "class"] | None = "class"
    validate_ref = lambda c, r: isclass(r)

    def __init__(self, __obj: Type[T]) -> None:
        """
        Args:
            __obj: The structured type to create adapter from

        Raises:
            ValueError: If object is of not class type
        """
        if not self.validate_ref(__obj):
            raise ValueError(f"Expected a {self.ref_type!r}")
        self.__ref = __obj
        self.__name = self.__ref.__name__

        from docstring_parser import parse
        from pydantic.type_adapter import TypeAdapter

        self.__docstring = parse(self.__ref.__doc__ or "")
        self.__adapter = TypeAdapter(self.__ref)
        self.__description = build_description(self.__docstring)
        self.__definition: str | None = None
        self.__schema: Dict[str, Any] | None = None

    def __call__(self, *args: Any, **kwargs: Any) -> T:
        return self.__ref(*args, **kwargs)

    @property
    def name(self) -> str:
        return self.__name

    @property
    def description(self) -> str | None:
        return self.__description or None

    @property
    def docstring(self) -> Docstring:
        return self.__docstring

    @property
    def definition(self) -> str:
        """Code definition"""
        if self.__definition is None:
            self.__definition = get_code_definition(self.__ref, type=self.ref_type)
        return self.__definition

    @property
    def schema(self) -> Dict[str, Any]:
        """Schema in standard format."""
        if self.__schema is None:
            base_schema = self.__adapter.json_schema()
            if "$defs" in base_schema:
                from jsonref import replace_refs  # type: ignore[import-untyped]

                base_schema = replace_refs(base_schema, lazy_load=False)
                _ = base_schema.pop("$defs", None)
            self.__schema = update_object_schema(self.__ref, base_schema, self.__docstring)
        return self.__schema

    def validate_input(self, input: Dict[str, Any] | str, *, strict: bool | None = None, **kwargs: Any) -> T:  # noqa: A002
        """
        Validate the input against the schema.

        Args:
            input: The dictionary object or JSON string to validate.
            strict: Whether to perform strict validation.
            kwargs: Additional keyword arguments to pass to the TypeAdapter's validate method.

        Returns:
            The validated input.
        """
        if isinstance(input, str):
            return self.__adapter.validate_json(input, strict=strict, **kwargs)
        return self.__adapter.validate_python(input, strict=strict, **kwargs)

definition: str property

Code definition

schema: Dict[str, Any] property

Schema in standard format.

validate_input

Validate the input against the schema.

Parameters:

Name Type Description Default
input Dict[str, Any] | str

The dictionary object or JSON string to validate.

 required
strict bool | None

Whether to perform strict validation.

 None
kwargs Any

Additional keyword arguments to pass to the TypeAdapter's validate method.

 {}

Returns:

Type Description
T

The validated input.
Source code in orchestr8/adapter/struct.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def validate_input(self, input: Dict[str, Any] | str, *, strict: bool | None = None, **kwargs: Any) -> T:  # noqa: A002
    """
    Validate the input against the schema.

    Args:
        input: The dictionary object or JSON string to validate.
        strict: Whether to perform strict validation.
        kwargs: Additional keyword arguments to pass to the TypeAdapter's validate method.

    Returns:
        The validated input.
    """
    if isinstance(input, str):
        return self.__adapter.validate_json(input, strict=strict, **kwargs)
    return self.__adapter.validate_python(input, strict=strict, **kwargs)

FunctionAdapter

A specialized adapter for Python functions, extending StructAdapter to handle function-specific type hints.

Provides: - Function code definition extraction - Dynamic schema generation for function signatures. - Extraction of docstring metadata for populating schema. - Automatic validation of python or JsON input against the schema.

Enables schema-driven representations of functions for function calling, serialization, and documentation purposes.

from orchestr8.adapter.function import FunctionAdapter

def create_user(id: str, role: Literal["admin", "user"]) -> bool:
    """
    Creates a new user with the given ID and role.

    :param id: The ID of the user to create.
    :param role: The role of the user to create.
    """

# Create adapter
fn = FunctionAdapter(create_user)

# Normal invocation
fn(id="user123", role="user")

# Access function properties
print(f"Function name: {fn.name}")
print(f"Function description: {fn.description}")
print(f"Function docstring: {fn.docstring}")
print(f"Function definition: {fn.definition}")
print(f"Function schema: {fn.schema}")
print(f"Function OpenAI schema: {fn.openai_schema}")
print(f"Function Anthropic schema: {fn.anthropic_schema}")
print(f"Function Gemini schema: {fn.gemini_schema}")

# Validate python input
fn.validate_input({"id": "user123", "role": "user"})

# Validate JSON input
fn.validate_input('{"id": "user123", "role": "user"}')
Source code in orchestr8/adapter/function.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class FunctionAdapter(StructAdapter[T_Retval], Generic[P_Spec, T_Retval]):  # type: ignore[misc]
    """
    A specialized adapter for Python functions, extending StructAdapter to handle function-specific type hints.

    Provides:
    - Function code definition extraction
    - Dynamic schema generation for function signatures.
    - Extraction of docstring metadata for populating schema.
    - Automatic validation of python or JsON input against the schema.

    Enables schema-driven representations of functions for function calling, serialization, and documentation purposes.

    ```python
    from orchestr8.adapter.function import FunctionAdapter

    def create_user(id: str, role: Literal["admin", "user"]) -> bool:
        \"""
        Creates a new user with the given ID and role.

        :param id: The ID of the user to create.
        :param role: The role of the user to create.
        \"""

    # Create adapter
    fn = FunctionAdapter(create_user)

    # Normal invocation
    fn(id="user123", role="user")

    # Access function properties
    print(f"Function name: {fn.name}")
    print(f"Function description: {fn.description}")
    print(f"Function docstring: {fn.docstring}")
    print(f"Function definition: {fn.definition}")
    print(f"Function schema: {fn.schema}")
    print(f"Function OpenAI schema: {fn.openai_schema}")
    print(f"Function Anthropic schema: {fn.anthropic_schema}")
    print(f"Function Gemini schema: {fn.gemini_schema}")

    # Validate python input
    fn.validate_input({"id": "user123", "role": "user"})

    # Validate JSON input
    fn.validate_input('{"id": "user123", "role": "user"}')
    ```
    """

    ref_type = "function"
    validate_ref = lambda c, r: isfunction(r)

    def __init__(self, __fn: Callable[P_Spec, T_Retval]) -> None:  # type: ignore[valid-type]
        """
        Args:
            __fn: The function to create adapter from

        Raises:
            ValueError: If __fn is not a valid function
        """
        super().__init__(__fn)  # type: ignore[arg-type]

    def __call__(self, *args: P_Spec.args, **kwargs: P_Spec.kwargs) -> T_Retval:  # type: ignore[name-defined]
        return super().__call__(*args, **kwargs)

@adapt decorator

Create an adapter from a function or structured type such as Pydantic Model, TypedDict, dataclass, etc.

from orchestr8.adapter import adapt
from typing_extensions import TypedDict
from pydantic import BaseModel

@adapt
def add(a: int, b: int) -> int:
    """
    Add two numbers together.

    :param a: The first number to add.
    :param b: The second number to add.
    """
    return a + b

@adapt
class User(TypedDict):
    id: str
    role: Literal["admin", "user"]
Source code in orchestr8/adapter/__init__.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def adapt(__obj: Any) -> Any:
    """
    Create an adapter from a function or structured type such as
    Pydantic Model, TypedDict, dataclass, etc.

    ```python
    from orchestr8.adapter import adapt
    from typing_extensions import TypedDict
    from pydantic import BaseModel

    @adapt
    def add(a: int, b: int) -> int:
        \"""
        Add two numbers together.

        :param a: The first number to add.
        :param b: The second number to add.
        \"""
        return a + b

    @adapt
    class User(TypedDict):
        id: str
        role: Literal["admin", "user"]
    ```
    """
    if isfunction(__obj):
        return FunctionAdapter(__obj)
    elif isclass(__obj):
        return StructAdapter(__obj)
    raise ValueError(f"Expected a function or class, got {type(__obj)} type")