Transforms the input schema into the specified output format, retaining only 'type', 'enum', and 'description' fields in the final dict.
(bundle_id, plugin_id, plugin_description, input_schema: Dict)
| 29 | |
| 30 | |
| 31 | def transform_input_schema(bundle_id, plugin_id, plugin_description, input_schema: Dict) -> ChatCompletionFunction: |
| 32 | """ |
| 33 | Transforms the input schema into the specified output format, |
| 34 | retaining only 'type', 'enum', and 'description' fields in the final dict. |
| 35 | """ |
| 36 | from app.services.tool import i18n_text |
| 37 | |
| 38 | # Initialize the base structure of the output schema |
| 39 | output_schema = { |
| 40 | "name": "object", |
| 41 | "properties": {}, |
| 42 | "required": [], |
| 43 | } |
| 44 | |
| 45 | for key, value in input_schema.items(): |
| 46 | # Copy only 'type', 'enum', and 'description' fields to the new schema |
| 47 | filtered_value = {k: v for k, v in value.items() if k in ["type", "enum", "description"]} |
| 48 | output_schema["properties"][key] = filtered_value |
| 49 | |
| 50 | if "array" in filtered_value.get("type", ""): |
| 51 | array_type = filtered_value["type"].replace("_array", "") |
| 52 | output_schema["properties"][key]["type"] = "array" |
| 53 | output_schema["properties"][key]["items"] = {"type": array_type} |
| 54 | |
| 55 | if filtered_value.get("description"): |
| 56 | # handle i18n |
| 57 | output_schema["properties"][key]["description"] = i18n_text(bundle_id, value["description"], "en") |
| 58 | |
| 59 | # If the original field is required, add it to the 'required' list in the output schema |
| 60 | if value.get("required"): |
| 61 | output_schema["required"].append(key) |
| 62 | |
| 63 | function = ChatCompletionFunction( |
| 64 | name=plugin_id, |
| 65 | description=i18n_text(bundle_id, plugin_description, "en"), |
| 66 | parameters=output_schema, |
| 67 | ) |
| 68 | return function |
| 69 | |
| 70 | |
| 71 | class Plugin(BaseModel): |
no test coverage detected