Function run_plugin

backend/app/services/tool/plugin/plugin.py:18–86

Run a plugin :param bundle_instance_id: the bundle ID :param plugin_id: the action ID :param parameters: the parameters for the API call :return: the response of the API call

(
    bundle_instance_id: str,
    plugin_id: str,
    parameters: Dict,
)

async def run_plugin(
    bundle_instance_id: str,
    plugin_id: str,
    parameters: Dict,
) -> Dict:
    """
    Run a plugin
    :param bundle_instance_id: the bundle ID
    :param plugin_id: the action ID
    :param parameters: the parameters for the API call
    :return: the response of the API call
    """
    bundle_instance: BundleInstance = await bundle_instance_ops.get(
        bundle_instance_id=bundle_instance_id,
    )

    try:
        async with aiohttp.ClientSession() as session:
            response = await session.post(
                f"{CONFIG.TASKINGAI_PLUGIN_URL}/v1/execute",
                json={
                    "bundle_id": bundle_instance.bundle_id,
                    "plugin_id": plugin_id,
                    "input_params": parameters,
                    "encrypted_credentials": bundle_instance.encrypted_credentials,
                    "project_id": CONFIG.PROJECT_ID
                },
            )

            bytes_read = 0
            max_size = 64 * 1024
            data_chunks = []

            # check the size of the response
            async for chunk in response.content.iter_any():
                bytes_read += len(chunk)
                if bytes_read > max_size:
                    raise ClientResponseError(
                        response.request_info, response.history, message="Response too large", status=response.status
                    )
                data_chunks.append(chunk)

            data_bytes = b"".join(data_chunks)
            try:
                # Assuming the response is JSON and decode here
                data_dict = json.loads(data_bytes.decode("utf-8"))
            except json.JSONDecodeError:
                # Handle non-JSON response or decode error
                return {"status": 500, "data": {"error": "Failed to decode the plugin response"}}

            response_wrapper = ResponseWrapper(response.status, data_dict)

            if response.status == 200:
                data = response_wrapper.json().get("data")
                return {"status": data["status"], "data": data["data"]}

            return {"status": response.status, "data": response_wrapper.json().get("error")}