Render a JSON-schema-ish description for resolver instructions. Cases handled: 1. ``Annotated[X, ...]`` — peel via ``__metadata__`` and recurse on X. 2. Plain ``BaseModel`` subclass — use ``model_json_schema()``. 3. ``Union[...]`` / ``T | U`` — recurse on each variant; emit
(t: Any)
| 160 | |
| 161 | def _strip_optional(anno: Any) -> Any: |
| 162 | """Peel ``T | None`` / ``Optional[T]`` to return the inner T.""" |
| 163 | origin = get_origin(anno) |
| 164 | if origin in (Union, types.UnionType): |
| 165 | non_none = [a for a in get_args(anno) if a is not type(None)] |
| 166 | if len(non_none) == 1: |
| 167 | return non_none[0] |
| 168 | return anno |
| 169 | |
| 170 | |
| 171 | def _pydantic_schema_str(t: Any) -> str: |
| 172 | """Render a JSON-schema-ish description for resolver instructions. |
| 173 | |
| 174 | Cases handled: |
| 175 | |
| 176 | 1. ``Annotated[X, ...]`` — peel via ``__metadata__`` and recurse on X. |
| 177 | 2. Plain ``BaseModel`` subclass — use ``model_json_schema()``. |
| 178 | 3. ``Union[...]`` / ``T | U`` — recurse on each variant; emit |
| 179 | ``{"oneOf": [...]}``. |
| 180 | 4. Anything else — fall back to ``repr`` so the resolver still gets |
| 181 | *some* hint. Should not happen for the supported flows. |
| 182 | """ |
| 183 | if hasattr(t, "__metadata__"): |
| 184 | inner = get_args(t)[0] |
| 185 | return _pydantic_schema_str(inner) |
| 186 | |
| 187 | if isinstance(t, type) and hasattr(t, "model_json_schema"): |
| 188 | try: |
| 189 | return json.dumps(t.model_json_schema(), indent=2) |
| 190 | except Exception: |
| 191 | # Some Pydantic models (e.g. those holding callable fields |
| 192 | # like ``ModelSettings`` from the agents SDK) cannot render |
| 193 | # a full JSON schema. Fall back to a name+fields summary |
| 194 | # so the resolver still has something to work with. |
| 195 | fields = { |
| 196 | name: repr(field.annotation) |
| 197 | for name, field in t.model_fields.items() |
| 198 | } |
| 199 | return json.dumps({"title": t.__name__, "fields": fields}, indent=2) |
| 200 | |
| 201 | origin = get_origin(t) |
| 202 | if origin in (Union, types.UnionType): |
| 203 | variants = get_args(t) |
| 204 | schemas = [ |
no outgoing calls