39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
from backend.tools.api.http_request_handler import http_request
|
|
from inspect import Signature, Parameter
|
|
from pydantic import BaseModel
|
|
from pydantic_core import PydanticUndefined
|
|
|
|
def create_tool(route: str, schema: type[BaseModel], remove_keys: list[str] | None = None):
|
|
"""Creates an MCP tool that calls an API route with Pydantic params."""
|
|
async def tool(**kwargs):
|
|
resolved_route = route.format(**kwargs)
|
|
filtered_kwargs = {k: v for k, v in kwargs.items() if f"{{{k}}}" not in route}
|
|
|
|
if not filtered_kwargs:
|
|
return await http_request(route=resolved_route, params={}, remove_keys=remove_keys)
|
|
|
|
parsed = schema(**filtered_kwargs)
|
|
return await http_request(
|
|
route=resolved_route,
|
|
params=parsed.model_dump(exclude_none=True),
|
|
remove_keys=remove_keys
|
|
)
|
|
|
|
params = []
|
|
for field_name, field_info in schema.model_fields.items():
|
|
default = field_info.default
|
|
params.append(
|
|
Parameter(
|
|
name=field_name,
|
|
kind=Parameter.KEYWORD_ONLY,
|
|
annotation=field_info.annotation,
|
|
default=default if default is not PydanticUndefined else Parameter.empty,
|
|
))
|
|
|
|
tool.__name__ = schema.__name__
|
|
tool.__doc__ = schema.__doc__
|
|
tool.__signature__ = Signature(params)
|
|
tool.__annotations__ = {f: i.annotation for f, i in schema.model_fields.items()}
|
|
|
|
return tool
|