Pydantic动态模型构建：运行时创建参数校验模型-深圳市維司達科技有限公司

摘要：MCP工具的参数结构各不相同，如何在运行时生成Pydantic校验模型？本文深入解析browser-use webui中的create_tool_param_model函数，讲解动态模型构建、类型递归解析和嵌套Schema处理。

一、问题背景：MCP工具的参数多样性

不同MCP Server提供的工具有截然不同的参数结构：

// 天气工具{"city":"string","days":"integer"}// 数据库工具{"query":"string","params":["array","of","strings"]}// 文件工具{"path":"string","content":"string","options":{"overwrite":"boolean","encoding":"string"}}

如果为每个工具手写Pydantic模型，维护成本极高。browser-use的解决方案是运行时动态创建模型。

二、动态模型创建入门

Pydantic提供了create_model函数，可以在运行时构建模型类：

frompydanticimportBaseModel,Field,create_model# 静态定义（传统方式）classWeatherInput(BaseModel):city:str=Field(description="城市名称")days:int=Field(default=7,ge=1,le=14)# 动态创建（等效于上面）WeatherInputDynamic=create_model('WeatherInput',city=(str,Field(description="城市名称")),days=(int,Field(default=7,ge=1,le=14)),)

三、browser-use的完整实现

3.1 主函数：从JSON Schema到Pydantic模型

# src/utils/mcp_client.pydefcreate_tool_param_model(tool:BaseTool)->Type[BaseModel]:""" 从LangChain工具的JSON Schema动态创建Pydantic参数模型 支持特性： - 基础类型映射（string/int/float/bool） - 数组类型（List[item_type]） - 嵌套对象（递归创建子模型） - 枚举类型（动态Enum类） - 联合类型（Union[A, B]） - 必填/选填字段 """json_schema=tool.args_schemaifjson_schemaisnotNoneand'properties'injson_schema:params={}required_fields=set(json_schema.get('required',[]))forprop_name,prop_detailsinjson_schema['properties'].items():# 递归解析字段类型field_type=resolve_type(prop_details,f"{tool.name}_{prop_name}")is_required=prop_nameinrequired_fields# 提取默认值和描述default_value=prop_details.get('default',...ifis_requiredelseNone)description=prop_details.get('description','')# 组装Field参数field_kwargs={'default':default_value}ifdescription:field_kwargs['description']=description# 数值约束if'minimum'inprop_details:field_kwargs['ge']=prop_details['minimum']if'maximum'inprop_details:field_kwargs['le']=prop_details['maximum']if'minLength'inprop_details:field_kwargs['min_length']=prop_details['minLength']if'maxLength'inprop_details:field_kwargs['max_length']=prop_details['maxLength']if'pattern'inprop_details:field_kwargs['pattern']=prop_details['pattern']params[prop_name]=(field_type,Field(**field_kwargs))# 以ActionModel为基类创建returncreate_model(f'{tool.name}_parameters',__base__=ActionModel,**params,)

3.2 类型解析器：递归处理复杂Schema

defresolve_type(prop_details:Dict[str,Any],prefix:str="")->Any:"""递归解析JSON Schema类型为Python/Pydantic类型"""# 基础类型映射表type_mapping={'string':str,'integer':int,'number':float,'boolean':bool,'array':List,'object':Dict,'null':type(None),}# 处理格式化字符串（日期、邮箱、URL等）ifprop_details.get('type')=='string'and'format'inprop_details:format_mapping={'date-time':datetime,'date':date,'time':time,'email':str,'uri':str,'url':str,'uuid':uuid.UUID,'binary':bytes,}returnformat_mapping.get(prop_details['format'],str)# 处理枚举类型：动态创建Enum类if'enum'inprop_details:enum_values=prop_details['enum']enum_dict={}fori,vinenumerate(enum_values):ifisinstance(v,str):key=v.upper().replace(' ','_').replace('-','_')ifnotkey.isidentifier():key=f"VALUE_{i}"else:key=f"VALUE_{i}"enum_dict[key]=vreturnEnum(f"{prefix}_Enum",enum_dict)# 处理数组类型：递归解析item类型ifprop_details.get('type')=='array'and'items'inprop_details:item_type=resolve_type(prop_details['items'],f"{prefix}_item")returnList[item_type]# 处理对象类型：递归创建嵌套模型ifprop_details.get('type')=='object'and'properties'inprop_details:nested_params={}fornested_name,nested_detailsinprop_details['properties'].items():nested_type=resolve_type(nested_details,f"{prefix}_{nested_name}")required_fields=prop_details.get('required',[])is_required=nested_nameinrequired_fields default_value=nested_details.get('default',...ifis_requiredelseNone)description=nested_details.get('description','')field_kwargs={'default':default_value}ifdescription:field_kwargs['description']=description nested_params[nested_name]=(nested_type,Field(**field_kwargs))nested_model=create_model(f"{prefix}_Model",**nested_params)returnnested_model# 处理联合类型（oneOf/anyOf）if'oneOf'inprop_detailsor'anyOf'inprop_details:union_schema=prop_details.get('oneOf')orprop_details.get('anyOf')union_types=[resolve_type(t,f"{prefix}_{i}")fori,tinenumerate(union_schema)]ifunion_types:returnUnion.__getitem__(tuple(union_types))returnAny# 默认返回基础类型returntype_mapping.get(prop_details.get('type'),Any)

四、实战案例：解析复杂工具Schema

4.1 输入Schema

{"type":"object","properties":{"query":{"type":"string","description":"SQL查询语句"},"params":{"type":"array","items":{"type":"string"},"description":"查询参数"},"mode":{"type":"string","enum":["read","write","admin"],"default":"read"},"timeout":{"type":"integer","minimum":1,"maximum":300,"default":30}},"required":["query"]}

4.2 生成的Pydantic模型

# 动态生成的等效代码classQueryMode_Enum(str,Enum):READ="read"WRITE="write"ADMIN="admin"classSqlQuery_parameters(BaseModel):query:str=Field(description="SQL查询语句")params:Optional[List[str]]=Field(default=None,description="查询参数")mode:QueryMode_Enum=Field(default=QueryMode_Enum.READ,description="")timeout:int=Field(default=30,ge=1,le=300,description="")