Skip to content

inspection

Class info

🛈 DocStrings

call_with_context 

call_with_context(func: Callable[..., T], context: AgentContext[Any], **kwargs: Any) -> T

Call function with appropriate context injection.

Handles: - Simple functions - Bound methods - Functions expecting AgentContext - Functions expecting context data

Source code in src/llmling_agent/utils/inspection.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def call_with_context[T](
    func: Callable[..., T],
    context: AgentContext[Any],
    **kwargs: Any,
) -> T:
    """Call function with appropriate context injection.

    Handles:
    - Simple functions
    - Bound methods
    - Functions expecting AgentContext
    - Functions expecting context data
    """
    from llmling_agent.agent import AgentContext

    if inspect.ismethod(func):
        if get_argument_key(func, AgentContext):
            return func(context)  # type: ignore[no-any-return]
        return func()  # type: ignore[no-any-return]
    if get_argument_key(func, AgentContext):
        return func(context, **kwargs)
    return func(context.data)

dataclasses_no_defaults_repr 

dataclasses_no_defaults_repr(self: Any) -> str

Exclude fields with values equal to the field default.

Source code in src/llmling_agent/utils/inspection.py 
33
34
35
36
37
38
39
40
def dataclasses_no_defaults_repr(self: Any) -> str:
    """Exclude fields with values equal to the field default."""
    kv_pairs = (
        f"{f.name}={getattr(self, f.name)!r}"
        for f in fields(self)
        if f.repr and getattr(self, f.name) != f.default
    )
    return f"{self.__class__.__qualname__}({', '.join(kv_pairs)})"

execute async 

execute(
    func: Callable[..., Awaitable[T]], *args: Any, use_thread: bool = False, **kwargs: Any
) -> T

execute(
    func: Callable[..., T | Awaitable[T]], *args: Any, use_thread: bool = False, **kwargs: Any
) -> T

execute(
    func: Callable[..., T | Awaitable[T]], *args: Any, use_thread: bool = False, **kwargs: Any
) -> T

Execute callable, handling both sync and async cases.

Source code in src/llmling_agent/utils/inspection.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
async def execute[T](
    func: Callable[..., T | Awaitable[T]],
    *args: Any,
    use_thread: bool = False,
    **kwargs: Any,
) -> T:
    """Execute callable, handling both sync and async cases."""
    if inspect.iscoroutinefunction(func):
        return await func(*args, **kwargs)  # type: ignore[no-any-return]

    if use_thread:
        result = await asyncio.to_thread(func, *args, **kwargs)
    else:
        result = func(*args, **kwargs)

    if inspect.iscoroutine(result) or inspect.isawaitable(result):
        return await result

    return result

get_argument_key 

get_argument_key(
    func: Callable[..., Any],
    arg_type: type | str | UnionType | Sequence[type | str | UnionType],
    include_return: bool = False,
) -> Literal[False] | str

Check if function has any argument of specified type(s) and return the key.

Parameters:

Name Type Description Default
func Callable[..., Any]

Function to check

 required
arg_type type | str | UnionType | Sequence[type | str | UnionType]

Type(s) to look for. Can be: - Single type (int, str, etc) - Union type (int | str) - Type name as string - Sequence of the above

 required
include_return bool

Whether to also check return type annotation

 False

Examples:

>>> def func(x: int | str, y: list[int]): ...
>>> get_argument_key(func, int | str)  # Returns 'x'
>>> get_argument_key(func, int)        # Returns 'x'
>>> get_argument_key(func, list)       # Returns 'y'
>>> get_argument_key(func, float)      # Returns False
>>> get_argument_key(func, (int, str)) # Returns 'x'

Returns:

Type Description
Literal[False] | str

Parameter name (str) if a matching argument is found, False otherwise.
Literal[False] | str

Only checks the origin type for generics, not type arguments
Literal[False] | str

(e.g., list[int] matches 'list', but does not match 'int').
Source code in src/llmling_agent/utils/inspection.py 
 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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def get_argument_key(
    func: Callable[..., Any],
    arg_type: type | str | UnionType | Sequence[type | str | UnionType],
    include_return: bool = False,
) -> Literal[False] | str:
    """Check if function has any argument of specified type(s) and return the key.

    Args:
        func: Function to check
        arg_type: Type(s) to look for. Can be:
            - Single type (int, str, etc)
            - Union type (int | str)
            - Type name as string
            - Sequence of the above
        include_return: Whether to also check return type annotation

    Examples:
        >>> def func(x: int | str, y: list[int]): ...
        >>> get_argument_key(func, int | str)  # Returns 'x'
        >>> get_argument_key(func, int)        # Returns 'x'
        >>> get_argument_key(func, list)       # Returns 'y'
        >>> get_argument_key(func, float)      # Returns False
        >>> get_argument_key(func, (int, str)) # Returns 'x'

    Returns:
        Parameter name (str) if a matching argument is found, False otherwise.
        Only checks the origin type for generics, not type arguments
        (e.g., list[int] matches 'list', but does not match 'int').
    """
    # Convert target type(s) to set of normalized strings
    if isinstance(arg_type, Sequence) and not isinstance(arg_type, str | bytes):
        target_types = {_type_to_string(t) for t in arg_type}
    else:
        target_types = {_type_to_string(arg_type)}

    # Get type hints including return type if requested
    hints = get_type_hints(func, include_extras=True)
    if not include_return:
        hints.pop("return", None)

    # Check each parameter's type annotation
    for key, param_type in hints.items():
        # Handle type aliases
        if isinstance(param_type, TypeAliasType):
            param_type = param_type.__value__

        # Check for direct match
        if _type_to_string(param_type) in target_types:
            return key

        # Handle Union types (both | and Union[...])
        origin = get_origin(param_type)
        if origin is Union or origin is UnionType:
            union_members = get_args(param_type)
            # Check each union member
            if any(_type_to_string(t) in target_types for t in union_members):
                return key
            # Also check if the complete union type matches
            if _type_to_string(param_type) in target_types:
                return key

        # Handle generic types (list[str], dict[str, int], etc)
        # Only check the origin type (e.g., list), not the arguments
        # This avoids matching nested contexts like RunContext[AgentContext]
        if origin is not None and _type_to_string(origin) in target_types:
            return key

        # if origin is not None:
        #     # Check if the generic type (e.g., list) matches
        #     if _type_to_string(origin) in target_types:
        #         return key
        #     # Check type arguments (e.g., str in list[str])
        #     args = get_args(param_type)
        #     if any(_type_to_string(arg) in target_types for arg in args):
        #         return key

    return False

has_return_type 

has_return_type(
    func: Callable[..., Any], expected_type: type[T]
) -> TypeGuard[Callable[..., T | Awaitable[T]]]

Check if a function has a specific return type annotation.

Parameters:

Name Type Description Default
func Callable[..., Any]

Function to check

 required
expected_type type[T]

The type to check for

 required

Returns:

Type Description
TypeGuard[Callable[..., T | Awaitable[T]]]

True if function returns the expected type (or Awaitable of it)
Source code in src/llmling_agent/utils/inspection.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def has_return_type[T](  # noqa: PLR0911
    func: Callable[..., Any],
    expected_type: type[T],
) -> TypeGuard[Callable[..., T | Awaitable[T]]]:
    """Check if a function has a specific return type annotation.

    Args:
        func: Function to check
        expected_type: The type to check for

    Returns:
        True if function returns the expected type (or Awaitable of it)
    """
    hints = get_type_hints(func)
    if "return" not in hints:
        return False

    return_type = hints["return"]

    # Handle direct match
    if return_type is expected_type:
        return True

    # Handle TypeAliases
    if isinstance(return_type, TypeAliasType):
        return_type = return_type.__value__

    # Handle Union types (including Optional)
    origin = get_origin(return_type)
    args = get_args(return_type)

    if origin is Union or origin is UnionType:
        # Check each union member
        def check_type(t: Any) -> bool:
            return has_return_type(lambda: t, expected_type)

        return any(check_type(arg) for arg in args)

    # Handle Awaitable/Coroutine types
    if origin is not None and inspect.iscoroutinefunction(func):
        # For async functions, check the first type argument
        if args:
            # Recursively check the awaited type
            return has_return_type(lambda: args[0], expected_type)
        return False

    # Handle generic types (like list[str], etc)
    if origin is not None:
        return origin is expected_type

    return False

is_async_callable 

is_async_callable(obj: Any) -> Any

Correctly check if a callable is async.

This function was copied from Starlette: https://github.com/encode/starlette/blob/78da9b9e218ab289117df7d62aee200ed4c59617/starlette/_utils.py#L36-L40

Source code in src/llmling_agent/utils/inspection.py 
161
162
163
164
165
166
167
168
169
170
171
172
def is_async_callable(obj: Any) -> Any:
    """Correctly check if a callable is async.

    This function was copied from Starlette:
    https://github.com/encode/starlette/blob/78da9b9e218ab289117df7d62aee200ed4c59617/starlette/_utils.py#L36-L40
    """
    while isinstance(obj, functools.partial):
        obj = obj.func

    return inspect.iscoroutinefunction(obj) or (
        callable(obj) and inspect.iscoroutinefunction(obj.__call__)
    )

validate_import 

validate_import(module_path: str, extras_name: str) -> None

Check existence of module, showing helpful error if not installed.

Source code in src/llmling_agent/utils/inspection.py 
269
270
271
272
273
274
275
276
def validate_import(module_path: str, extras_name: str) -> None:
    """Check existence of module, showing helpful error if not installed."""
    if not find_spec(module_path):
        msg = f"""
Optional dependency {module_path!r} not found.
Install with: pip install {PACKAGE_NAME}[{extras_name}]
"""
        raise ImportError(msg.strip())