Skip to content

griptape_cloud_structure_run_driver

GriptapeCloudStructureRunDriver

Bases: BaseStructureRunDriver

Source code in griptape/drivers/structure_run/griptape_cloud_structure_run_driver.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@define
class GriptapeCloudStructureRunDriver(BaseStructureRunDriver):
    base_url: str = field(default="https://cloud.griptape.ai", kw_only=True)
    api_key: str = field(kw_only=True)
    headers: dict = field(
        default=Factory(lambda self: {"Authorization": f"Bearer {self.api_key}"}, takes_self=True),
        kw_only=True,
    )
    structure_id: str = field(kw_only=True)
    structure_run_wait_time_interval: int = field(default=2, kw_only=True)
    structure_run_max_wait_time_attempts: int = field(default=20, kw_only=True)
    async_run: bool = field(default=False, kw_only=True)

    def try_run(self, *args: BaseArtifact) -> BaseArtifact | InfoArtifact:
        from requests import Response, post

        url = urljoin(self.base_url.strip("/"), f"/api/structures/{self.structure_id}/runs")

        env_vars = [{"name": key, "value": value, "source": "manual"} for key, value in self.env.items()]

        response: Response = post(
            url,
            json={"args": [arg.value for arg in args], "env_vars": env_vars},
            headers=self.headers,
        )
        response.raise_for_status()
        response_json = response.json()

        if self.async_run:
            return InfoArtifact("Run started successfully")
        else:
            return self._get_structure_run_result(response_json["structure_run_id"])

    def _get_structure_run_result(self, structure_run_id: str) -> BaseArtifact | InfoArtifact:
        url = urljoin(self.base_url.strip("/"), f"/api/structure-runs/{structure_run_id}")

        result = self._get_structure_run_result_attempt(url)
        status = result["status"]

        wait_attempts = 0
        while (
            status not in ("SUCCEEDED", "FAILED", "ERROR", "CANCELLED")
            and wait_attempts < self.structure_run_max_wait_time_attempts
        ):
            # wait
            time.sleep(self.structure_run_wait_time_interval)
            wait_attempts += 1
            result = self._get_structure_run_result_attempt(url)
            status = result["status"]

        if wait_attempts >= self.structure_run_max_wait_time_attempts:
            raise Exception(f"Failed to get Run result after {self.structure_run_max_wait_time_attempts} attempts.")

        if status != "SUCCEEDED":
            raise Exception(f"Run failed with status: {status}")

        if "output" in result:
            return BaseArtifact.from_dict(result["output"])
        else:
            return InfoArtifact("No output found in response")

    def _get_structure_run_result_attempt(self, structure_run_url: str) -> Any:
        from requests import Response, get

        response: Response = get(structure_run_url, headers=self.headers)
        response.raise_for_status()

        return response.json()

api_key: str = field(kw_only=True) class-attribute instance-attribute

async_run: bool = field(default=False, kw_only=True) class-attribute instance-attribute

base_url: str = field(default='https://cloud.griptape.ai', kw_only=True) class-attribute instance-attribute

headers: dict = field(default=Factory(lambda self: {'Authorization': f'Bearer {self.api_key}'}, takes_self=True), kw_only=True) class-attribute instance-attribute

structure_id: str = field(kw_only=True) class-attribute instance-attribute

structure_run_max_wait_time_attempts: int = field(default=20, kw_only=True) class-attribute instance-attribute

structure_run_wait_time_interval: int = field(default=2, kw_only=True) class-attribute instance-attribute

try_run(*args)

Source code in griptape/drivers/structure_run/griptape_cloud_structure_run_driver.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def try_run(self, *args: BaseArtifact) -> BaseArtifact | InfoArtifact:
    from requests import Response, post

    url = urljoin(self.base_url.strip("/"), f"/api/structures/{self.structure_id}/runs")

    env_vars = [{"name": key, "value": value, "source": "manual"} for key, value in self.env.items()]

    response: Response = post(
        url,
        json={"args": [arg.value for arg in args], "env_vars": env_vars},
        headers=self.headers,
    )
    response.raise_for_status()
    response_json = response.json()

    if self.async_run:
        return InfoArtifact("Run started successfully")
    else:
        return self._get_structure_run_result(response_json["structure_run_id"])