Skip to content

func_metadata

StrictJsonSchema

Bases: GenerateJsonSchema

A JSON schema generator that raises exceptions instead of emitting warnings.

This is used to detect non-serializable types during schema generation.

Source code in src/mcp/server/mcpserver/utilities/func_metadata.py 
36
37
38
39
40
41
42
43
44
class StrictJsonSchema(GenerateJsonSchema):
    """A JSON schema generator that raises exceptions instead of emitting warnings.

    This is used to detect non-serializable types during schema generation.
    """

    def emit_warning(self, kind: JsonSchemaWarningKind, detail: str) -> None:
        # Raise an exception instead of emitting a warning
        raise ValueError(f"JSON schema warning: {kind} - {detail}")

ArgModelBase

Bases: BaseModel

A model representing the arguments to a function.

Source code in src/mcp/server/mcpserver/utilities/func_metadata.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class ArgModelBase(BaseModel):
    """A model representing the arguments to a function."""

    def model_dump_one_level(self) -> dict[str, Any]:
        """Return a dict of the model's fields, one level deep.

        That is, sub-models etc are not dumped - they are kept as Pydantic models.
        """
        kwargs: dict[str, Any] = {}
        for field_name, field_info in self.__class__.model_fields.items():
            value = getattr(self, field_name)
            # Use the alias if it exists, otherwise use the field name
            output_name = field_info.alias if field_info.alias else field_name
            kwargs[output_name] = value
        return kwargs

    model_config = ConfigDict(arbitrary_types_allowed=True)

model_dump_one_level 

model_dump_one_level() -> dict[str, Any]

Return a dict of the model's fields, one level deep.

That is, sub-models etc are not dumped - they are kept as Pydantic models.

Source code in src/mcp/server/mcpserver/utilities/func_metadata.py 
50
51
52
53
54
55
56
57
58
59
60
61
def model_dump_one_level(self) -> dict[str, Any]:
    """Return a dict of the model's fields, one level deep.

    That is, sub-models etc are not dumped - they are kept as Pydantic models.
    """
    kwargs: dict[str, Any] = {}
    for field_name, field_info in self.__class__.model_fields.items():
        value = getattr(self, field_name)
        # Use the alias if it exists, otherwise use the field name
        output_name = field_info.alias if field_info.alias else field_name
        kwargs[output_name] = value
    return kwargs

FuncMetadata

Bases: BaseModel

Source code in src/mcp/server/mcpserver/utilities/func_metadata.py 
 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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
class FuncMetadata(BaseModel):
    arg_model: Annotated[type[ArgModelBase], WithJsonSchema(None)]
    output_schema: dict[str, Any] | None = None
    output_model: Annotated[type[BaseModel], WithJsonSchema(None)] | None = None
    wrap_output: bool = False

    def validate_arguments(self, arguments_to_validate: dict[str, Any]) -> dict[str, Any]:
        """Validate raw arguments into a one-level kwargs dict (no function call).

        Used to feed resolver dependency injection the validated tool arguments
        before the tool function itself runs.
        """
        arguments_pre_parsed = self.pre_parse_json(arguments_to_validate)
        arguments_parsed_model = self.arg_model.model_validate(arguments_pre_parsed)
        return arguments_parsed_model.model_dump_one_level()

    async def call_fn_with_arg_validation(
        self,
        fn: Callable[..., Any | Awaitable[Any]],
        fn_is_async: bool,
        arguments_to_validate: dict[str, Any],
        arguments_to_pass_directly: dict[str, Any] | None,
        pre_validated: dict[str, Any] | None = None,
    ) -> Any:
        """Call the given function with arguments validated and injected.

        Arguments are first attempted to be parsed from JSON, then validated against
        the argument model, before being passed to the function. Pass `pre_validated`
        (the output of `validate_arguments`) to reuse an earlier validation pass -
        validating twice can re-run `default_factory`/stateful validators and hand the
        function different values than a caller already observed.
        """
        # Copy so a caller-provided `pre_validated` dict is never mutated in place.
        arguments_parsed_dict = dict(
            pre_validated if pre_validated is not None else self.validate_arguments(arguments_to_validate)
        )

        arguments_parsed_dict |= arguments_to_pass_directly or {}

        if fn_is_async:
            return await fn(**arguments_parsed_dict)
        else:
            return await anyio.to_thread.run_sync(functools.partial(fn, **arguments_parsed_dict))

    def convert_result(self, result: Any) -> CallToolResult | InputRequiredResult:
        """Convert a function call result into a `CallToolResult`.

        An `InputRequiredResult` is passed through unchanged so the multi-round
        flow surfaces on the wire as `resultType: "input_required"` rather than
        being JSON-dumped into a text block.

        Note: we build unstructured content here **even though the lowlevel server
        tool call handler provides generic backwards compatibility serialization of
        structured content**. This is for MCPServer backwards compatibility: we need to
        retain MCPServer's ad hoc conversion logic for constructing unstructured output
        from function return values, whereas the lowlevel server simply serializes
        the structured output.
        """
        if isinstance(result, InputRequiredResult):
            return result
        if isinstance(result, CallToolResult):
            if self.output_schema is not None:
                assert self.output_model is not None, "Output model must be set if output schema is defined"
                self.output_model.model_validate(result.structured_content)
            return result

        unstructured_content = _convert_to_content(result)

        if self.output_schema is None:
            return CallToolResult(content=unstructured_content)

        if self.wrap_output:
            result = {"result": result}

        assert self.output_model is not None, "Output model must be set if output schema is defined"
        validated = self.output_model.model_validate(result)
        structured_content = validated.model_dump(mode="json", by_alias=True)

        return CallToolResult(content=unstructured_content, structured_content=structured_content)

    def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]:
        """Pre-parse data from JSON.

        Return a dict with the same keys as input but with values parsed from JSON
        if appropriate.

        This is to handle cases like `["a", "b", "c"]` being passed in as JSON inside
        a string rather than an actual list. Claude Desktop is prone to this - in fact
        it seems incapable of NOT doing this. For sub-models, it tends to pass
        dicts (JSON objects) as JSON strings, which can be pre-parsed here.
        """
        new_data = data.copy()  # Shallow copy

        # Build a mapping from input keys (including aliases) to field info
        key_to_field_info: dict[str, FieldInfo] = {}
        for field_name, field_info in self.arg_model.model_fields.items():
            # Map both the field name and its alias (if any) to the field info
            key_to_field_info[field_name] = field_info
            if field_info.alias:
                key_to_field_info[field_info.alias] = field_info

        for data_key, data_value in data.items():
            if data_key not in key_to_field_info:
                continue

            field_info = key_to_field_info[data_key]
            if isinstance(data_value, str) and field_info.annotation is not str:
                try:
                    pre_parsed = json.loads(data_value)
                except json.JSONDecodeError:
                    continue  # Not JSON - skip
                if isinstance(pre_parsed, str | int | float):
                    # This is likely that the raw value is e.g. `"hello"` which we
                    # Should really be parsed as '"hello"' in Python - but if we parse
                    # it as JSON it'll turn into just 'hello'. So we skip it.
                    continue
                new_data[data_key] = pre_parsed
        assert new_data.keys() == data.keys()
        return new_data

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )

validate_arguments 

validate_arguments(
    arguments_to_validate: dict[str, Any],
) -> dict[str, Any]

Validate raw arguments into a one-level kwargs dict (no function call).

Used to feed resolver dependency injection the validated tool arguments before the tool function itself runs.

Source code in src/mcp/server/mcpserver/utilities/func_metadata.py 
72
73
74
75
76
77
78
79
80
def validate_arguments(self, arguments_to_validate: dict[str, Any]) -> dict[str, Any]:
    """Validate raw arguments into a one-level kwargs dict (no function call).

    Used to feed resolver dependency injection the validated tool arguments
    before the tool function itself runs.
    """
    arguments_pre_parsed = self.pre_parse_json(arguments_to_validate)
    arguments_parsed_model = self.arg_model.model_validate(arguments_pre_parsed)
    return arguments_parsed_model.model_dump_one_level()

call_fn_with_arg_validation async 

call_fn_with_arg_validation(
    fn: Callable[..., Any | Awaitable[Any]],
    fn_is_async: bool,
    arguments_to_validate: dict[str, Any],
    arguments_to_pass_directly: dict[str, Any] | None,
    pre_validated: dict[str, Any] | None = None,
) -> Any

Call the given function with arguments validated and injected.

Arguments are first attempted to be parsed from JSON, then validated against the argument model, before being passed to the function. Pass pre_validated (the output of validate_arguments) to reuse an earlier validation pass - validating twice can re-run default_factory/stateful validators and hand the function different values than a caller already observed.

Source code in src/mcp/server/mcpserver/utilities/func_metadata.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
async def call_fn_with_arg_validation(
    self,
    fn: Callable[..., Any | Awaitable[Any]],
    fn_is_async: bool,
    arguments_to_validate: dict[str, Any],
    arguments_to_pass_directly: dict[str, Any] | None,
    pre_validated: dict[str, Any] | None = None,
) -> Any:
    """Call the given function with arguments validated and injected.

    Arguments are first attempted to be parsed from JSON, then validated against
    the argument model, before being passed to the function. Pass `pre_validated`
    (the output of `validate_arguments`) to reuse an earlier validation pass -
    validating twice can re-run `default_factory`/stateful validators and hand the
    function different values than a caller already observed.
    """
    # Copy so a caller-provided `pre_validated` dict is never mutated in place.
    arguments_parsed_dict = dict(
        pre_validated if pre_validated is not None else self.validate_arguments(arguments_to_validate)
    )

    arguments_parsed_dict |= arguments_to_pass_directly or {}

    if fn_is_async:
        return await fn(**arguments_parsed_dict)
    else:
        return await anyio.to_thread.run_sync(functools.partial(fn, **arguments_parsed_dict))

convert_result 

convert_result(
    result: Any,
) -> CallToolResult | InputRequiredResult

Convert a function call result into a CallToolResult.

An InputRequiredResult is passed through unchanged so the multi-round flow surfaces on the wire as resultType: "input_required" rather than being JSON-dumped into a text block.

Note: we build unstructured content here even though the lowlevel server tool call handler provides generic backwards compatibility serialization of structured content. This is for MCPServer backwards compatibility: we need to retain MCPServer's ad hoc conversion logic for constructing unstructured output from function return values, whereas the lowlevel server simply serializes the structured output.

Source code in src/mcp/server/mcpserver/utilities/func_metadata.py 
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
def convert_result(self, result: Any) -> CallToolResult | InputRequiredResult:
    """Convert a function call result into a `CallToolResult`.

    An `InputRequiredResult` is passed through unchanged so the multi-round
    flow surfaces on the wire as `resultType: "input_required"` rather than
    being JSON-dumped into a text block.

    Note: we build unstructured content here **even though the lowlevel server
    tool call handler provides generic backwards compatibility serialization of
    structured content**. This is for MCPServer backwards compatibility: we need to
    retain MCPServer's ad hoc conversion logic for constructing unstructured output
    from function return values, whereas the lowlevel server simply serializes
    the structured output.
    """
    if isinstance(result, InputRequiredResult):
        return result
    if isinstance(result, CallToolResult):
        if self.output_schema is not None:
            assert self.output_model is not None, "Output model must be set if output schema is defined"
            self.output_model.model_validate(result.structured_content)
        return result

    unstructured_content = _convert_to_content(result)

    if self.output_schema is None:
        return CallToolResult(content=unstructured_content)

    if self.wrap_output:
        result = {"result": result}

    assert self.output_model is not None, "Output model must be set if output schema is defined"
    validated = self.output_model.model_validate(result)
    structured_content = validated.model_dump(mode="json", by_alias=True)

    return CallToolResult(content=unstructured_content, structured_content=structured_content)

pre_parse_json 

pre_parse_json(data: dict[str, Any]) -> dict[str, Any]

Pre-parse data from JSON.

Return a dict with the same keys as input but with values parsed from JSON if appropriate.

This is to handle cases like ["a", "b", "c"] being passed in as JSON inside a string rather than an actual list. Claude Desktop is prone to this - in fact it seems incapable of NOT doing this. For sub-models, it tends to pass dicts (JSON objects) as JSON strings, which can be pre-parsed here.

Source code in src/mcp/server/mcpserver/utilities/func_metadata.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]:
    """Pre-parse data from JSON.

    Return a dict with the same keys as input but with values parsed from JSON
    if appropriate.

    This is to handle cases like `["a", "b", "c"]` being passed in as JSON inside
    a string rather than an actual list. Claude Desktop is prone to this - in fact
    it seems incapable of NOT doing this. For sub-models, it tends to pass
    dicts (JSON objects) as JSON strings, which can be pre-parsed here.
    """
    new_data = data.copy()  # Shallow copy

    # Build a mapping from input keys (including aliases) to field info
    key_to_field_info: dict[str, FieldInfo] = {}
    for field_name, field_info in self.arg_model.model_fields.items():
        # Map both the field name and its alias (if any) to the field info
        key_to_field_info[field_name] = field_info
        if field_info.alias:
            key_to_field_info[field_info.alias] = field_info

    for data_key, data_value in data.items():
        if data_key not in key_to_field_info:
            continue

        field_info = key_to_field_info[data_key]
        if isinstance(data_value, str) and field_info.annotation is not str:
            try:
                pre_parsed = json.loads(data_value)
            except json.JSONDecodeError:
                continue  # Not JSON - skip
            if isinstance(pre_parsed, str | int | float):
                # This is likely that the raw value is e.g. `"hello"` which we
                # Should really be parsed as '"hello"' in Python - but if we parse
                # it as JSON it'll turn into just 'hello'. So we skip it.
                continue
            new_data[data_key] = pre_parsed
    assert new_data.keys() == data.keys()
    return new_data

func_metadata 

func_metadata(
    func: Callable[..., Any],
    skip_names: Sequence[str] = (),
    structured_output: bool | None = None,
) -> FuncMetadata

Given a function, return metadata including a Pydantic model representing its signature.

The use case for this is 

meta = func_metadata(func)
validated_args = meta.arg_model.model_validate(some_raw_data_dict)
return func(**validated_args.model_dump_one_level())

critically it also provides a pre-parse helper to attempt to parse things from JSON.

Parameters:

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

The function to convert to a Pydantic model

 required
skip_names Sequence[str]

A list of parameter names to skip. These will not be included in the model.

 ()
structured_output bool | None

Controls whether the tool's output is structured or unstructured - If None, auto-detects based on the function's return type annotation - If True, creates a structured tool (return type annotation permitting) - If False, unconditionally creates an unstructured tool

If structured, creates a Pydantic model for the function's result based on its annotation. Supports various return types: - BaseModel subclasses (used directly) - Primitive types (str, int, float, bool, bytes, None) - wrapped in a model with a 'result' field - TypedDict - converted to a Pydantic model with same fields - Dataclasses and other annotated classes - converted to Pydantic models - Generic types (list, dict, Union, etc.) - wrapped in a model with a 'result' field

 None

Returns:

Type Description
FuncMetadata

A FuncMetadata object containing:
FuncMetadata
  • arg_model: A Pydantic model representing the function's arguments
FuncMetadata
  • output_model: A Pydantic model for the return type if the output is structured
FuncMetadata
  • wrap_output: Whether the function result needs to be wrapped in {"result": ...} for structured output.
Source code in src/mcp/server/mcpserver/utilities/func_metadata.py 
191
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
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def func_metadata(
    func: Callable[..., Any],
    skip_names: Sequence[str] = (),
    structured_output: bool | None = None,
) -> FuncMetadata:
    """Given a function, return metadata including a Pydantic model representing its signature.

    The use case for this is
    ```
    meta = func_metadata(func)
    validated_args = meta.arg_model.model_validate(some_raw_data_dict)
    return func(**validated_args.model_dump_one_level())
    ```

    **critically** it also provides a pre-parse helper to attempt to parse things from
    JSON.

    Args:
        func: The function to convert to a Pydantic model
        skip_names: A list of parameter names to skip. These will not be included in
            the model.
        structured_output: Controls whether the tool's output is structured or unstructured
            - If None, auto-detects based on the function's return type annotation
            - If True, creates a structured tool (return type annotation permitting)
            - If False, unconditionally creates an unstructured tool

            If structured, creates a Pydantic model for the function's result based on its annotation.
            Supports various return types:
            - BaseModel subclasses (used directly)
            - Primitive types (str, int, float, bool, bytes, None) - wrapped in a
                model with a 'result' field
            - TypedDict - converted to a Pydantic model with same fields
            - Dataclasses and other annotated classes - converted to Pydantic models
            - Generic types (list, dict, Union, etc.) - wrapped in a model with a 'result' field

    Returns:
        A FuncMetadata object containing:
        - arg_model: A Pydantic model representing the function's arguments
        - output_model: A Pydantic model for the return type if the output is structured
        - wrap_output: Whether the function result needs to be wrapped in `{"result": ...}` for structured output.
    """
    try:
        sig = inspect.signature(func, eval_str=True)
    except NameError as e:  # pragma: no cover
        # This raise could perhaps be skipped, and we (MCPServer) just call
        # model_rebuild right before using it 🤷
        raise InvalidSignature(f"Unable to evaluate type annotations for callable {func.__name__!r}") from e
    params = sig.parameters
    dynamic_pydantic_model_params: dict[str, Any] = {}
    for param in params.values():
        if param.name.startswith("_"):  # pragma: no cover
            raise InvalidSignature(f"Parameter {param.name} of {func.__name__} cannot start with '_'")
        if param.name in skip_names:
            continue

        annotation = param.annotation if param.annotation is not inspect.Parameter.empty else Any
        field_name = param.name
        field_kwargs: dict[str, Any] = {}
        field_metadata: list[Any] = []

        if param.annotation is inspect.Parameter.empty:
            field_metadata.append(WithJsonSchema({"title": param.name, "type": "string"}))
        # Check if the parameter name conflicts with BaseModel attributes
        # This is necessary because Pydantic warns about shadowing parent attributes
        if hasattr(BaseModel, field_name) and callable(getattr(BaseModel, field_name)):
            # Use an alias to avoid the shadowing warning
            field_kwargs["alias"] = field_name
            # Use a prefixed field name
            field_name = f"field_{field_name}"

        if param.default is not inspect.Parameter.empty:
            dynamic_pydantic_model_params[field_name] = (
                Annotated[(annotation, *field_metadata, Field(**field_kwargs))],
                param.default,
            )
        else:
            dynamic_pydantic_model_params[field_name] = Annotated[(annotation, *field_metadata, Field(**field_kwargs))]

    arguments_model = create_model(
        f"{func.__name__}Arguments",
        __base__=ArgModelBase,
        **dynamic_pydantic_model_params,
    )

    if structured_output is False:
        return FuncMetadata(arg_model=arguments_model)

    # set up structured output support based on return type annotation

    if sig.return_annotation is inspect.Parameter.empty and structured_output is True:
        raise InvalidSignature(f"Function {func.__name__}: return annotation required for structured output")

    try:
        inspected_return_ann = inspect_annotation(sig.return_annotation, annotation_source=AnnotationSource.FUNCTION)
    except ForbiddenQualifier as e:
        raise InvalidSignature(f"Function {func.__name__}: return annotation contains an invalid type qualifier") from e

    return_type_expr = inspected_return_ann.type

    # `AnnotationSource.FUNCTION` allows no type qualifier to be used, so `return_type_expr` is guaranteed to *not* be
    # unknown (i.e. a bare `Final`).
    assert return_type_expr is not UNKNOWN

    if _is_input_required_type(return_type_expr):
        # A tool annotated to return only InputRequiredResult never produces structured content.
        return FuncMetadata(arg_model=arguments_model)

    # The annotation fed to schema derivation. Starts as the raw return annotation (preserving any
    # Annotated[...] wrapper) and is narrowed below if InputRequiredResult arms are stripped.
    effective_annotation: Any = sig.return_annotation

    if is_union_origin(get_origin(return_type_expr)):
        args = get_args(return_type_expr)
        # InputRequiredResult is a control-flow signal, not data: strip it so the residual arms
        # drive schema derivation. convert_result short-circuits on an InputRequiredResult instance
        # before output validation, so the schema only ever sees the data arms at runtime.
        residual = tuple(a for a in args if not _is_input_required_type(a))
        if not residual:
            return FuncMetadata(arg_model=arguments_model)
        if len(residual) != len(args):
            # PEP 604 has no syntax for "union of a runtime tuple"; Union[...] is the only spelling.
            effective_annotation = residual[0] if len(residual) == 1 else Union[residual]  # noqa: UP007
            # Re-normalize so the residual is processed exactly as if it had been the declared
            # return annotation: unwraps a top-level Annotated[...] arm and re-derives metadata,
            # so the CallToolResult/BaseModel/TypedDict dispatch below sees the bare type.
            inspected_return_ann = inspect_annotation(effective_annotation, annotation_source=AnnotationSource.FUNCTION)
            return_type_expr = inspected_return_ann.type
        if len(residual) > 1 and any(
            isinstance(a, type) and issubclass(a, CallToolResult) for a in residual if a is not type(None)
        ):
            raise InvalidSignature(
                f"Function {func.__name__}: CallToolResult cannot be used in Union or Optional types. "
                "To return empty results, use: CallToolResult(content=[])"
            )

    original_annotation: Any
    # if the typehint is CallToolResult, the user either intends to return without validation
    # or they provided validation as Annotated metadata
    if isinstance(return_type_expr, type) and issubclass(return_type_expr, CallToolResult):
        if inspected_return_ann.metadata:
            return_type_expr = inspected_return_ann.metadata[0]
            if len(inspected_return_ann.metadata) >= 2:
                # Reconstruct the original annotation, by preserving the remaining metadata,
                # i.e. from `Annotated[CallToolResult, ReturnType, Gt(1)]` to
                # `Annotated[ReturnType, Gt(1)]`:
                original_annotation = Annotated[
                    (return_type_expr, *inspected_return_ann.metadata[1:])
                ]  # pragma: no cover
            else:
                # We only had `Annotated[CallToolResult, ReturnType]`, treat the original annotation
                # as being `ReturnType`:
                original_annotation = return_type_expr
        else:
            return FuncMetadata(arg_model=arguments_model)
    else:
        original_annotation = effective_annotation

    output_model, output_schema, wrap_output = _try_create_model_and_schema(
        original_annotation, return_type_expr, func.__name__
    )

    if output_model is None and structured_output is True:
        # Model creation failed or produced warnings - no structured output
        raise InvalidSignature(
            f"Function {func.__name__}: return type {return_type_expr} is not serializable for structured output"
        )

    return FuncMetadata(
        arg_model=arguments_model,
        output_schema=output_schema,
        output_model=output_model,
        wrap_output=wrap_output,
    )