Repository: Significant-Gravitas/Auto-GPT
File: autogpt_platform/backend/backend/util/service.py (Python)

Code before:
            orig_func = getattr(service_type, name, None)
            if orig_func is None:
                raise AttributeError(f"Method {name} not found in {service_type}")

            sig = inspect.signature(orig_func)
            ret_ann = sig.return_annotation
            if ret_ann != inspect.Signature.empty:
                expected_return = TypeAdapter(ret_ann)
            else:
                expected_return = None

            def method(*args, **kwargs) -> Any:
                if args:
                    arg_names = list(sig.parameters.keys())
                    if arg_names[0] in ("self", "cls"):
                        arg_names = arg_names[1:]
                    kwargs.update(dict(zip(arg_names, args)))
                result = self.client.call_method(name, **kwargs)
                if expected_return:
                    return expected_return.validate_python(result)
                return result

            return method

    client = cast(AS, DynamicClient())
    if client.health_check() != "OK":
        raise RuntimeError(f"Health check failed for {service_name}")

    return cast(AS, DynamicClient())


Review comment (bug):
I guess this works, but `"OK"` is a magic value that can't be differentiated by type from an unhealthy response. I'd suggest letting `health_check` raise an error on failure and treating an error-less call to `health_check` as healthy.

Code after:

class FastApiAppService(BaseAppService, ABC):
    fastapi_app: FastAPI

    def _create_fastapi_endpoint(self, func: Callable) -> Callable:
        """
        Generates a FastAPI endpoint for the given function, handling default and optional parameters.

        :param func: The original function (sync/async, bound or unbound)
        :return: A FastAPI endpoint function.
        """
        sig = inspect.signature(func)
        fields = {}

        is_bound_method = False
        for name, param in sig.parameters.items():
            if name in ("self", "cls"):
                is_bound_method = True
                continue

            # Use the provided annotation or fallback to str if not specified
            annotation = (
                param.annotation if param.annotation != inspect.Parameter.empty else str
            )

            # If a default value is provided, use it; otherwise, mark the field as required with '...'
            default = param.default if param.default != inspect.Parameter.empty else ...

            fields[name] = (annotation, default)

        # Dynamically create a Pydantic model for the request body
        RequestBodyModel = create_model("RequestBodyModel", **fields)
        f = func.__get__(self) if is_bound_method else func

        if asyncio.iscoroutinefunction(f):

            async def async_endpoint(body: RequestBodyModel):  # type: ignore #RequestBodyModel being variable
                try:
                    return await f(
                        **{name: getattr(body, name) for name in body.model_fields}
                    )
                except Exception as e:
                    logger.exception(f"Error in {func.__name__}: {e}")
                    raise HTTPException(status_code=500, detail=e)

            return async_endpoint
        else:

            def sync_endpoint(body: RequestBodyModel):  # type: ignore #RequestBodyModel being variable
                try:
                    return f(