Skip to main content

Python SDK

Roadmap

The FluidGrids Python SDK (fluidgrids-sdk) is not yet available on PyPI. This page describes the planned package. Until it ships, use the Node SDK, the CLI, or call the GraphQL API from Python with any HTTP client.

The FluidGrids Python SDK will provide async/await access to the GraphQL API.

Install​

Not available yet. Once it is published, the install will be:

pip install fluidgrids-sdk

The import name will be fluidgrids_sdk.

Environment selection​

The SDK endpoint defaults to production. Select a different environment by passing the appropriate endpoint in FluidGridsConfig:

EnvironmentEndpoint
prod (default)https://graphqlworkspaces.burdenoff.com/workspaces/graphql
alphahttps://alphagraphqlworkspaces.burdenoff.com/workspaces/graphql
localhttp://localhost:4003/workspaces/graphql

The endpoint comes from the endpoint= argument to FluidGridsConfig, or from the FLUIDGRIDS_ENDPOINT environment variable (the BURDENOFF_ENDPOINT alias also works). BURDENOFF_ENV is read only by the browser and VS Code extensions, not by this SDK.

Quick start​

import asyncio
from fluidgrids_sdk import FluidGridsSDK, FluidGridsConfig

async def main():
    config = FluidGridsConfig(
        endpoint="https://graphqlworkspaces.burdenoff.com/workspaces/graphql",
        access_token="your-token"
    )

    async with FluidGridsSDK(config) as sdk:
        workflows = await sdk.workflows.list()
        print(f"Found {workflows.total} workflows")

        for workflow in workflows.items:
            print(f"  - {workflow.name} ({workflow.key})")

asyncio.run(main())

Authentication​

API Key​

config = FluidGridsConfig(
    endpoint="https://graphqlworkspaces.burdenoff.com/workspaces/graphql",
    api_key="your-api-key"
)

JWT Token​

config = FluidGridsConfig(
    endpoint="https://graphqlworkspaces.burdenoff.com/workspaces/graphql",
    access_token="your-jwt-token"
)

Dynamic Token Update​

sdk = FluidGridsSDK(config)
sdk.set_tokens(access_token="new-token", refresh_token="refresh-token")
sdk.clear_tokens()

Workflow Automation​

Workflows​

from fluidgrids_sdk import CreateWorkflowInput, UpdateWorkflowInput, TriggerWorkflowInput, WorkflowStatus

# Note: there is no PUBLISHED status — publishing a workflow moves it to ACTIVE.
workflows = await sdk.workflows.list(limit=10, status=WorkflowStatus.ACTIVE)
workflow = await sdk.workflows.get_by_id("workflow-uuid")
workflow = await sdk.workflows.get_by_key("my-workflow", version="v1.0.0")
new_workflow = await sdk.workflows.create(CreateWorkflowInput(
    name="My Workflow",
    description="A test workflow",
    tags=["test"]
))
updated = await sdk.workflows.update("workflow-id", UpdateWorkflowInput(description="Updated description"))
await sdk.workflows.delete("workflow-id")
result = await sdk.workflows.trigger_by_key(
    key="my-workflow",
    input_data=TriggerWorkflowInput(input_data={"name": "John"}, is_async=True)
)
await sdk.workflows.publish("workflow-id", change_log="Release notes")
copy = await sdk.workflows.duplicate("workflow-id", name="Copy of Workflow")

Runs​

from fluidgrids_sdk import ListRunsOptions, LogLevel, RunStatus

runs = await sdk.runs.list(ListRunsOptions(
    workflow_key="my-workflow",
    status=RunStatus.COMPLETED,
    limit=20
))
run = await sdk.runs.get("run-id")
logs = await sdk.runs.get_logs("run-id", level=LogLevel.ERROR)
await sdk.runs.cancel("run-id")
await sdk.runs.pause("run-id")
await sdk.runs.resume("run-id")
await sdk.runs.retry("run-id")
completed_run = await sdk.runs.wait_for_completion(
    "run-id",
    poll_interval=2.0,
    timeout=60.0
)

Nodes​

nodes = await sdk.nodes.list(category="Integrations")
node = await sdk.nodes.get("node-id")
categories = await sdk.nodes.list_categories()
results = await sdk.nodes.search("http")

Credentials​

from fluidgrids_sdk import CreateCredentialInput

credentials = await sdk.credentials.list()
credential = await sdk.credentials.create(CreateCredentialInput(
    name="My API Key",
    type="api_key",
    data={"apiKey": "secret"}
))
result = await sdk.credentials.test("credential-id")
await sdk.credentials.delete("credential-id")

Webhooks​

from fluidgrids_sdk import CreateWebhookInput

webhooks = await sdk.webhooks.list(workflow_id="workflow-id")
webhook = await sdk.webhooks.create(CreateWebhookInput(
    name="My Webhook",
    workflow_id="workflow-id",
    method="POST"
))
await sdk.webhooks.activate("webhook-id")
await sdk.webhooks.deactivate("webhook-id")
await sdk.webhooks.delete("webhook-id")

Scheduled Jobs​

from fluidgrids_sdk import CreateScheduledJobInput

jobs = await sdk.scheduled_jobs.list()
job = await sdk.scheduled_jobs.create(CreateScheduledJobInput(
    name="Daily Report",
    workflow_id="workflow-id",
    cron_expression="0 9 * * *",
    timezone="America/New_York"
))
await sdk.scheduled_jobs.pause("job-id")
await sdk.scheduled_jobs.resume("job-id")
await sdk.scheduled_jobs.delete("job-id")

Dashboard​

stats = await sdk.dashboard.get_stats("7d")
print(f"Total workflows: {stats.total_workflows}")
print(f"Success rate: {stats.success_rate}%")
last_24h = await sdk.dashboard.get_last_24_hours()
last_7d = await sdk.dashboard.get_last_7_days()
last_30d = await sdk.dashboard.get_last_30_days()

Error Handling​

try:
    workflow = await sdk.workflows.get_by_id("non-existent")
except Exception as e:
    print(f"Error: {e}")

Next steps​

Last updated on