Skip to main content

Python SDK

The official Python SDK for FluidGrids provides async/await access to the full GraphQL API.

Install​

fluidgrids-sdk is not published to PyPI yet. Install it from the repository:

pip install "git+https://github.com/algoshred/fluidgrids-sdk-python.git"

Development install:

git clone https://github.com/algoshred/fluidgrids-sdk-python.git
cd fluidgrids-sdk-python
python -m venv venv && source venv/bin/activate
pip install -e ".[dev]"

Once published, the install becomes pip install fluidgrids-sdk. The import name is fluidgrids_sdk either way.

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

workflows = await sdk.workflows.list(limit=10, status=WorkflowStatus.PUBLISHED)
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, 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}")

Development​

python -m venv venv && source venv/bin/activate
pip install -e ".[dev]"
pytest
pytest --cov=fluidgrids_sdk
black src/ tests/
isort src/ tests/
flake8 src/ tests/
mypy src/

Source​

  • Repository: github.com/algoshred/fluidgrids-sdk-python
  • Package: fluidgrids-sdk on PyPI

Next steps​