CLIClient¶

Coming Soon

CLIClient is not yet publicly available. It will be released in an upcoming release. The documentation below is provided as a preview of upcoming capabilities. For code running inside the platform, use FlowSDK instead.

CLIClient is a synchronous HTTP client for invoking deployed hosted services and managing platform resources from scripts, notebooks, and automation tooling. It uses stored credentials from ~/.flow/config.json (created by flow-sdk login).

When to Use CLIClient

Use CLIClient when you need to call the platform from outside a running service -- local scripts, CI/CD pipelines, Jupyter notebooks, or integration tests. For code running inside a hosted service or code block, use FlowSDK instead.

Setup¶

Before using CLIClient, authenticate with the platform:

flow-sdk login --api-key fl_your_api_key --url https://api.flow.marut.cloud

This stores a JWT token and tenant context in ~/.flow/config.json.

Basic Usage¶

from flow_sdk.config import load_config
from flow_sdk.cli_client import CLIClient

config = load_config()
client = CLIClient(config)

Invoking Hosted Services¶

The primary use case for CLIClient is calling deployed hosted service endpoints.

`invoke_service()`¶

result = client.invoke_service(
    service_id="<service-uuid>",
    method="POST",
    path="/analyze",
    body={"text": "Analyze this document for sentiment."},
)
print(result)
# {"sentiment": "positive", "confidence": 0.92}

Parameters¶

Parameter	Type	Required	Description
`service_id`	`str`	Yes	UUID of the deployed hosted service
`method`	`str`	Yes	HTTP method (`GET`, `POST`, `PUT`, `DELETE`)
`path`	`str`	Yes	Endpoint path (e.g., `/analyze`, `/health`)
`body`	`dict \\| None`	No	JSON request body
`query`	`dict \\| None`	No	URL query parameters
`version`	`int \\| None`	No	Pin to a specific service version

Version Pinning¶

Pin requests to a specific version of a service:

# Call version 2 of the service
result = client.invoke_service(
    service_id="<service-uuid>",
    method="GET",
    path="/status",
    version=2,
)

Examples¶

POST with BodyGET with Query ParamsQuick Test Script

result = client.invoke_service(
    service_id="nlp-service-uuid",
    method="POST",
    path="/summarize",
    body={
        "text": "Long document content...",
        "max_length": 200,
    },
)
print(result["summary"])

records = client.invoke_service(
    service_id="data-service-uuid",
    method="GET",
    path="/records",
    query={"status": "active", "limit": "50"},
)
for record in records:
    print(record["name"])

"""Quick test script for a deployed HSL service."""
from flow_sdk.config import load_config
from flow_sdk.cli_client import CLIClient

config = load_config()
client = CLIClient(config)

# Health check
health = client.invoke_service(
    service_id="<service-uuid>",
    method="GET",
    path="/health",
)
print(f"Health: {health}")

# Functional test
result = client.invoke_service(
    service_id="<service-uuid>",
    method="POST",
    path="/process",
    body={"input": "test data"},
)
print(f"Result: {result}")

API Key Exchange¶

CLIClient can exchange an organization API key for a JWT access token:

from flow_sdk.cli_client import CLIClient

token = CLIClient.exchange_api_key(
    platform_url="https://api.flow.marut.cloud",
    api_key="fl_your_api_key",
    ttl_minutes=60,
)
print(f"Token: {token[:20]}...")

This is the same exchange that flow-sdk login performs internally.

Platform Management Methods¶

CLIClient also provides methods for platform resource management:

Canonical Namespaces

Use canonical namespaces only: client.component_deployments.*, client.workflow_deployments.*, client.rings.*, and client.hosted_services.service_tokens.*.

Deployment Rings¶

# List available deployment rings
rings = client.rings.list()
for ring in rings:
    print(f"{ring['name']} ({ring['id']})")

# Resolve a ring name to its UUID
ring_id = client.rings.resolve("dev")

Component Deployments¶

# Create a new deployment
deployment = client.component_deployments.create({
    "ring_id": ring_id,
    "deployment_notes": "release candidate",
    "components": [
        {
            "component_id": "<component-uuid>",
            "component_type": "codeblock",
        }
    ],
})

# Check deployment status
deployment_id = deployment["deployments"][0]["id"]
status = client.component_deployments.status(deployment_id)
print(f"Status: {status['status']}")

Component Publishing¶

# Publish a component to the catalog
component = client.create_component(
    component_type="codeblock",
    data={
        "name": "my-processor",
        "version": "1.0.0",
        "description": "Processes incoming data.",
    },
)

Organizations and Workspaces¶

# List accessible organizations
orgs = client.list_orgs()

# List workspaces within an organization
workspaces = client.list_workspaces(org_id="<org-uuid>")

Service Token Management¶

Manage HSL service tokens for programmatic service access:

# Mint a service token
token_result = client.hosted_services.service_tokens.mint(
    service_id="<service-uuid>",
    claims={"tenant_id": "t-123", "role": "reader"},
    ttl_minutes=120,
)
print(f"Token JTI: {token_result['jti']}")
print(f"Expires:   {token_result['expires_at']}")

# Revoke a specific token
client.hosted_services.service_tokens.revoke(
    service_id="<service-uuid>",
    jti=token_result["jti"],
    expires_at=token_result["expires_at"],
    reason="No longer needed",
)

# Revoke all tokens for a service
client.hosted_services.service_tokens.revoke_all(
    service_id="<service-uuid>",
    reason="Rotating credentials",
)

# List active revocations
revocations = client.hosted_services.service_tokens.list_revocations(
    service_id="<service-uuid>",
    include_expired=False,
)

Breaking Changes¶

The canonical namespace-only surface removes these non-canonical entry points:

CLIClient.mint_service_token(api_key, ...) removed.
Use client.hosted_services.service_tokens.mint(service_id=..., claims=..., ttl_minutes=...).
CLIClient.deployments.* removed.
Use client.component_deployments.*, client.workflow_deployments.*, and client.rings.*.
CLIRingsNamespace.create_ring(...) removed.
Use client.rings.create(...).

Connector DDL¶

Preview or execute DDL statements against connector instances:

# Preview DDL (dry run)
preview = client.execute_ddl(
    instance_id="<connector-instance-uuid>",
    sql="CREATE TABLE events (id TEXT, payload JSONB)",
    confirm=False,
)
print(preview)  # Shows what would happen

# Execute DDL for real
result = client.execute_ddl(
    instance_id="<connector-instance-uuid>",
    sql="CREATE TABLE events (id TEXT, payload JSONB)",
    confirm=True,
    reason="Setting up event storage",
)

Error Handling¶

CLIClient maps HTTP status codes to typed exceptions:

from flow_sdk.cli_client import (
    CLIClient,
    CLIClientError,
    CLIAuthenticationError,
    CLIPermissionError,
    CLINotFoundError,
)

try:
    result = client.invoke_service(
        service_id="nonexistent-uuid",
        method="GET",
        path="/health",
    )
except CLIAuthenticationError:
    print("Token expired. Run: flow-sdk login --api-key ... --url ...")
except CLIPermissionError:
    print("You do not have access to this service.")
except CLINotFoundError:
    print("Service or endpoint not found.")
except CLIClientError as e:
    print(f"API error: {e}")

Exception Hierarchy¶

Exception	HTTP Status	Description
`CLIAuthenticationError`	401	Token expired or invalid
`CLIPermissionError`	403	Insufficient permissions
`CLINotFoundError`	404	Service, endpoint, or resource not found
`CLIClientError`	Other	Base exception for all other errors

Credential Verification¶

Verify that stored credentials are still valid:

try:
    client.verify_credentials()
    print("Credentials are valid.")
except CLIAuthenticationError:
    print("Credentials expired. Please re-authenticate.")