Metadata-Version: 2.4
Name: automio
Version: 0.3.0
Summary: Automio API
Home-page: 
Author: OpenAPI Generator community
Author-email: team@openapitools.org
Keywords: OpenAPI,OpenAPI-Generator,Automio API
Description-Content-Type: text/markdown
Requires-Dist: urllib3<3.0.0,>=2.1.0
Requires-Dist: python-dateutil>=2.8.2
Requires-Dist: pydantic>=2
Requires-Dist: typing-extensions>=4.7.1
Requires-Dist: python-dotenv>=1.1.0
Requires-Dist: requests>=2.32.0
Dynamic: author
Dynamic: author-email
Dynamic: description
Dynamic: description-content-type
Dynamic: keywords
Dynamic: requires-dist
Dynamic: summary

﻿# Automio Python Client - API Documentation

A Python client for Automio API that enables seamless integration with various APIs and services.

> **For development documentation** (versioning, publishing, etc.), see [README.dev.md](README.dev.md)

## Installation

```bash
pip install automio
```

## Quick Start

### Environment Variables

Create a `.env` file in your project root for configuration:

```env
# Cache Service (when available)
CACHE_SERVICE_URL=https://code-cache.automio.ai
CACHE_SERVICE_TOKEN=your_cache_token

# Redis Configuration (for local development)
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=optional_password
ORGANIZATION_ID=optional_local_org_id

# Automio API Configuration
API_BASE_URL=https://automio.ai
APP_TOKEN=your_api_token
APP_ID=your_app_id
ACTION_ID=optional_runtime_action_id

# Optional outbound proxy (standard Python proxy variables)
HTTPS_PROXY=http://proxy.internal:3128
NO_PROXY=localhost,127.0.0.1

# Task Management
TASK_ID=optional_custom_task_id
```

### Usage Example

```python
from automio.wrapper import API

api = API()
```

The generated Automio API client honors `HTTP_PROXY`, `HTTPS_PROXY`, and
`NO_PROXY`. An explicit `Configuration.proxy` value takes precedence; set it
to an empty string to disable environment proxy discovery for that client.

### Request retries

By default, the wrapper retries transient HTTP failures such as upstream `401`, upstream `5xx`, and connection drops with exponential backoff.

Disable wrapper-level retries when you want the first response immediately, for example in diagnostic actions:

```python
from automio.wrapper import API

api = API(retry_requests=False)
```

## Actions

### Run an action

```python
from automio.wrapper import API

api = API()

result = api.actions.run(
    action_slug="my-action-slug",
    user_input={  # optional
        "date_from": "2026-01-01",
        "date_to": "2026-01-31",
    },
)

print(result.run_id, result.task_id, result.run_number)
```

Method signature:

```python
api.actions.run(
    action_slug: str | int | None = None,
    selected_rows: list[str] | None = None,
    selected_all: bool | None = None,
    filter_params: dict | None = None,
    user_input: dict | None = None,
    connection_id: str | None = None,
    language: str | None = None,
)
```

`user_input` is optional, but important and frequently used for runtime parameters from the action form (`USER_INPUT` in action code).

## Files

### Upload an application file

```python
from pathlib import Path
from automio.wrapper import API

api = API()

result = api.files.upload(
    file=Path("report.txt"),
    path="exports/report.txt",
    overwrite=True,
)

print(result.public_url)
print(result.object_key)
```

Method signature:

```python
api.files.upload(
    file: bytes | bytearray | str | os.PathLike | tuple[str, bytes],
    path: str | None = None,
    overwrite: bool = False,
    app_id: int | str | None = None,
    organization_id: int | str | None = None,
)
```

Notes:

- If `app_id` is omitted, the wrapper uses `APP_ID` from the environment.
- Passing `Path("local-file.txt")` is supported and will upload that local file.
- `path` is the deterministic logical path used to derive the final S3 object key.
- Re-uploading the same `app_id + organization context + path` requires `overwrite=True`.
- `organization_id` enables delegated upload in `app_action` context — see [Delegated file upload in app_action](#delegated-file-upload-in-app_action).

### Upload an application file from URL

```python
from automio.wrapper import API

api = API()

result = api.files.upload_from_url(
    url="https://example.com/reports/report.txt",
    path="imports/report.txt",
    overwrite=True,
)

print(result.public_url)
print(result.object_key)
```

Method signature:

```python
api.files.upload_from_url(
    url: str,
    path: str | None = None,
    overwrite: bool = False,
    app_id: int | str | None = None,
    organization_id: int | str | None = None,
)
```

Notes:

- If `app_id` is omitted, the wrapper uses `APP_ID` from the environment.
- `path` works the same as in `files.upload(...)`.
- If `path` is omitted, the backend derives the file name from the URL path, ignoring query params.
- The backend downloads the remote file first and only uploads to S3 after a successful full download.
- `organization_id` enables delegated upload in `app_action` context — see [Delegated file upload in app_action](#delegated-file-upload-in-app_action).

### Resolve a file URL

```python
from automio.wrapper import API

api = API()

result = api.files.get_metadata("exports/report.txt")
print(result.public_url)
```

Method signature:

```python
api.files.get_metadata(
    path: str,
    app_id: int | str | None = None,
    organization_id: int | str | None = None,
)
```

### Download stored file bytes

Application storage URLs returned by `api.files.*` use an automatically
allowlisted, application-specific S3 hostname. You can download one with a
normal HTTP client:

```python
import requests

from automio.wrapper import API

api = API()

metadata = api.files.get_metadata("exports/report.txt")
response = requests.get(metadata.public_url, timeout=(10, 30))
response.raise_for_status()
content = response.content

print(len(content))
```

Only exact storage hostnames generated for existing Automio applications are
allowed. The shared S3 provider domain and arbitrary buckets are not
allowlisted.

Alternatively, use the authenticated Automio API to validate the file against
the current application and organization before returning its bytes:

```python
content = api.files.download_binary(metadata.public_url)
```

Method signature:

```python
api.files.download_binary(
    public_url: str,
    app_id: int | str | None = None,
    organization_id: int | str | None = None,
    timeout: float | tuple[float, float] | None = None,
) -> bytes
```

`public_url` must be an absolute HTTP(S) URL previously returned for a file in
the same application and organization context. The backend verifies the file
reference and returns its bytes through the Automio API.

### Delete a file

```python
from automio.wrapper import API

api = API()

result = api.files.delete("exports/report.txt")
print(result.detail)
print(result.object_key)
```

Method signature:

```python
api.files.delete(
    path: str,
    app_id: int | str | None = None,
    organization_id: int | str | None = None,
)
```

Notes:

- If `app_id` is omitted, the wrapper uses `APP_ID` from the environment.
- Deletion is scoped by the current backend organization context, the same as upload and download.

### Delegated file operations in app_action

In `app_action` context, you can operate on files on behalf of another organization by passing `organization_id`. This applies to `upload`, `upload_from_url`, `get_metadata`, `download_binary`, and `delete`. The backend validates that:

- The action exists, is of type `app_action`, and is public.
- The calling organization owns both the action and the application.
- The target organization has an active, authorized App-Token for the application.
- The target organization has the application activated.

```python
from pathlib import Path
from automio.wrapper import API

api = API()

# Upload on behalf of organization 42
result = api.files.upload(
    file=Path("report.pdf"),
    path="exports/report.pdf",
    overwrite=True,
    organization_id=42,
)
print(result.public_url)

# Upload from URL on behalf of organization 42
result = api.files.upload_from_url(
    url="https://example.com/data.csv",
    path="imports/data.csv",
    overwrite=True,
    organization_id=42,
)
print(result.public_url)

# Resolve file URL on behalf of organization 42
result = api.files.get_metadata("exports/report.pdf", organization_id=42)
print(result.public_url)

# Download file bytes on behalf of organization 42
content = api.files.download_binary(
    result.public_url,
    organization_id=42,
)
print(len(content))

# Delete file on behalf of organization 42
result = api.files.delete("exports/report.pdf", organization_id=42)
print(result.detail)
```

This pattern is available only in `app_action`. For `user_action`, `table_action`, `trigger_action`, and `webhook_action`, `organization_id` is ignored and file operations are scoped to `request.user.organization`.

## Table Tags (Tables API)

Use table tag helpers from `api.tables` to list, create, get, update, and delete tags.

### List tags

```python
from automio.wrapper import API

api = API()

tags = api.tables.list_tags("threads")
for tag in tags:
    print(tag.id, tag.name, tag.color)
```

Method signature:

```python
api.tables.list_tags(table_name: str) -> list
```

### Create tag

```python
from automio.wrapper import API

api = API()

new_tag = api.tables.create_tag("threads", "Important")
print(new_tag.id, new_tag.name)
```

Method signature:

```python
api.tables.create_tag(table_name: str, tag_name: str)
```

### Get tag

```python
from automio.wrapper import API

api = API()

tag = api.tables.get_tag(123)
print(tag.id, tag.name, tag.description, tag.color)
```

Method signature:

```python
api.tables.get_tag(tag_id: int | str)
```

### Update tag

```python
from automio.wrapper import API

api = API()

updated_tag = api.tables.update_tag(
    123,
    tag_name="Important",
    description="Orders requiring manual review",
    color="#971a1a",
)
print(updated_tag.id, updated_tag.name, updated_tag.description, updated_tag.color)
```

Method signature:

```python
api.tables.update_tag(
    tag_id: int | str,
    tag_name: str | None = None,
    description: str | None = None,
    color: str | None = None,
)
```

### Delete tag

```python
from automio.wrapper import API

api = API()

deleted = api.tables.delete_tag(123)
print(deleted)
```

Method signature:

```python
api.tables.delete_tag(tag_id: int | str) -> bool
```

Color note:

- Tag color is a HEX string, for example `#971a1a`.
- When creating a tag, if color is not provided, a random color is assigned automatically.

## AI Router

Use `api.ai` for application AI instead of calling provider URLs directly.
The router selects the app-specific provider and credential source configured by
the organization running the action, injects credentials server-side, records
usage, and returns a billing breakdown.

### Provider-neutral generation

```python
data, response = api.ai.generate(
    model="automio-balanced",
    input="Return the order number and total as JSON.",
    response_schema={
        "type": "object",
        "properties": {
            "order_number": {"type": "string"},
            "total": {"type": "number"},
        },
        "required": ["order_number", "total"],
        "additionalProperties": False,
    },
)

print(data["text"], data.get("json"))
print(response.provider, response.model, response.credential_source)
print(response.billing.ai_usage_credits_charged)
print(response.billing.call_credits_charged)
print(response.billing.total_credits_charged)
```

`generate()` requires a global platform alias: `automio-fast`,
`automio-balanced`, or `automio-pro`. The same action code follows the
calling organization's OpenAI, Anthropic, or Gemini application route.
Provider-neutral `generate_stream()` uses the same route and aliases.
Provider-native `call()` and `stream()` remain available for advanced
provider-specific requests.

### Provider-native calls

Use `call()` for allowlisted provider-native parameters that are not part of
the normalized generation API:

```python
data, response = api.ai.call(
    model="your-enabled-model",
    endpoint="/responses",
    request_id="invoice-summary-123",
    payload={
        "input": "Summarize this invoice.",
        "reasoning": {"effort": "medium"},
    },
)
```

This example applies to an OpenAI-configured route. `endpoint` and `payload`
must match that route. Use an allowlisted endpoint path, the top-level `model`
argument, and `api.ai.stream(...)` for streaming instead of full URLs, custom
authentication headers, `payload["model"]`, or `payload["stream"]`.
Managed routes additionally reject raw provider resource IDs, hosted tools,
media, prompt-cache modifiers, and other fields not covered by token billing.
Use a connected integration route for supported provider-native fields.
`request_id` is an optional deduplication and correlation key. The first
accepted request claims it within the current application; reusing it returns
`ai_idempotency_conflict` (`409`) and does not replay the response body. After
an ambiguous client timeout, check AI Router usage history for that ID before
deciding whether to issue a new request ID.

### Streaming

```python
for event in api.ai.generate_stream(
    model="automio-balanced",
    input="Write a short status update.",
):
    if event["type"] == "text_delta":
        print(event["data"]["text"], end="", flush=True)
    elif event["type"] == "done":
        billing = event.get("metadata", {}).get("billing", {})
        print("\nCredits:", billing.get("total_credits_charged"))
    elif event["type"] == "error":
        print("\nAI error:", event["data"]["detail"])
```

The iterator yields the same normalized `text_delta`, `usage`, `done`, and
`error` events for OpenAI, Anthropic, and Gemini. The terminal `done.data`
contains complete text, optional parsed JSON, and the finish reason;
`done.metadata` contains final usage and billing.

Use `api.ai.stream(...)` only when you need raw provider events and can supply
the configured provider's allowlisted endpoint and native payload.

### Image generation and editing

Generate images through the configured route without selecting a provider or
provider endpoint:

```python
generated, generation_response = api.ai.images.generate(
    model="automio-image-balanced",
    prompt="Create a clean product hero image.",
    request_id="product-hero-v3",
)
```

Image calls use the provider and credentials configured in AI Router's
**Images** section, independently of text routing. Generation requires one of
`automio-image-fast`, `automio-image-balanced`, or `automio-image-pro`; AI
Router resolves that alias to the selected provider model. Direct provider
model names are not accepted.

Image edits use a dedicated multipart helper and the normal application route:

```python
data, response = api.ai.images.edit(
    image=["images/product.png", "images/style.webp"],
    mask="images/mask.png",
    prompt="Replace the background while preserving the product.",
    output_format="webp",
    output_compression=85,
    size="1536x1024",
    request_id="product-background-v3",
)
```

The example includes an image mask, so it requires a regular OpenAI API route,
not Codex, even though the model is selected automatically. Gemini routes
support image editing with the common `image` and `prompt` fields. Binary
masks, `partial_images`, and OpenAI output options are rejected on Gemini.
Anthropic does not provide image generation. Gemini image editing currently
requires a connected Gemini account, and the total inline request must be
smaller than 20 MB.

`image` accepts a path, bytes, an open binary file, or up to 16 such values in
a list/tuple. Use `api.ai.images.stream_edit(...)` for provider-native image
events followed by normalized usage and billing; `partial_images=1..3` is
OpenAI-only. `/images/edits` is intentionally unavailable through JSON
`call()`.

### Video generation

Video generation is asynchronous and uses the application's separate
**Videos** route:

```python
generation, response = api.ai.videos.generate(
    model="automio-video-balanced",
    prompt="A paper plane flying above Warsaw at sunrise.",
)
completed = api.ai.videos.wait(generation["id"], timeout=600)
if completed["status"] == "succeeded":
    video_bytes = api.ai.videos.download(completed["id"])
```

Use `api.ai.videos.retrieve(id)` for one status refresh. Gemini, Kling, and
Seedance are supported through connected provider accounts. Submission incurs
one standard Automio call fee; polling and download do not.

Provider capabilities differ:

| Operation | OpenAI | Anthropic | Gemini |
| --- | --- | --- | --- |
| `generate`, `generate_stream` | Yes | Yes | Yes |
| `images.generate` | Yes | No | Yes, connected account |
| `images.edit`, `images.stream_edit` | Yes | No | Yes, connected account |

| Video operation | Gemini | Kling | Seedance |
| --- | --- | --- | --- |
| `videos.generate`, `retrieve`, `wait`, `download` | Yes | Yes | Yes |
Do not retry capability or endpoint errors by
silently selecting another provider. Fix an invalid operation or endpoint in
application code, or ask the organization owner to select a compatible route.
Codex reports narrower limitations with specific `ai_codex_*` capability codes.

### Billing

- An Automio-managed key charges the measured provider usage as
  `ai_usage_credits_charged`.
- A connected provider integration keeps the existing standard flat call
  charge as `call_credits_charged` for every provider attempt, including a
  provider error; Automio does not additionally charge its token cost.
- `total_credits_charged` is always the sum of those two fields.
- All credit values are decimal strings. Do not convert them to binary floats
  for accounting.
Non-streaming Router failures and requests rejected before an SSE stream starts
raise `automio.AIRouterError`. After a stream starts, failures arrive as
`{"type": "error", "data": ...}` events. `AIRouterError` exposes `code`,
`detail`, `hint`, `retryable`, `provider`, `credential_source`, `request_id`,
`upstream_status`, and `status_code`; route and request metadata can be `None`
when the failure happens before route resolution or usage-record creation.
Stable errors include
`ai_route_not_configured`, `ai_capability_unavailable`,
`ai_insufficient_credits`, and `ai_app_budget_exceeded`.

Use `code` and `retryable` for control flow and `request_id` for correlation.
Do not parse `detail`. When `credential_source == "integration"`, the
organization is using its own connected provider account and `detail` contains
the provider's error message, limited to 4,000 characters. For
`credential_source == "automio"` or `"codex_user"`, `detail` is a normalized
Automio message. Provider response bodies, request payloads, authorization
headers, and credentials are never exposed. A provider message may mention
request data, so map `code` to an application-owned message before displaying
an error to an end user.

```python
from automio import AIRouterError

try:
    data, response = api.ai.generate(
        model="automio-balanced",
        input="Summarize this order.",
    )
except AIRouterError as error:
    print(error.code, error.detail, error.request_id)
    if error.retryable:
        schedule_bounded_retry()
    else:
        handle_permanent_failure(error.code)
```

## Core Methods

### GraphQL (Tables API)

Use `GraphQLRequest` together with `api.tables.apps_graphql_query(...)` to run queries and mutations against dynamic app tables (Hasura-backed). This is how the trigger scripts operate:

- Queries are scoped by `APP_ID` and table name (e.g., `app_{APP_ID}_orders`).
- Variables are passed via `GraphQLRequest(query=..., variables=...)`.
- Responses include `.data` and `.errors`; always check `.errors` before using data.
- Mutations use `insert_app_{APP_ID}_{TABLE_NAME}` or `update_app_{APP_ID}_{TABLE_NAME}` with `where` clauses and `_set` updates.
- UI, `user_action`, `table_action`, and `trigger_action` calls are scoped to the current runtime organization by the backend Hasura role.
- Public `app_action` code must pass `organization_id` when reading or writing one target organization's rows. The SDK sends the delegated runtime headers; it does not rewrite the GraphQL `where` clause.
- An `app_action` call without `organization_id` keeps creator-wide visibility and should be used only for app-owner discovery reads across activated organizations.

#### Delegated table access in `app_action`

Use explicit delegation for every tenant-specific read or mutation:

```python
result = api.tables.apps_graphql_query(
    app=APP_ID,
    table="orders",
    organization_id=target_organization_id,
    graph_ql_request=GraphQLRequest(query=query, variables=variables),
)
```

The backend verifies the public `app_action`, its owner, the runtime App-Token,
and the target organization's active application token/subscription. It then
executes GraphQL using the target organization's Hasura scope. Never put
`organization_id` in `_set` or `update_columns`; it is an immutable,
platform-managed tenant key.

#### Example: Query a single order by `orderId`

```python
from typing import Any, Dict
from automio import GraphQLRequest
from automio.wrapper import API

api = API()
APP_ID = 130
ORDERS_TABLE_NAME = "orders"

order_query = f"""query GetOrder($order_id: String!, $organization_id: Int) {{
    app_{APP_ID}_{ORDERS_TABLE_NAME}(
        where: {{orderId: {{_eq: $order_id}}, organization_id: {{_eq: $organization_id}}}},
        limit: 1
    ) {{
        id
        orderId
        buyerLogin
    }}
}}"""

variables: Dict[str, Any] = {"order_id": "eba6bc90-faab-11f0-96f9-c518c5ba21fb", "organization_id": 2}
request = GraphQLRequest(query=order_query, variables=variables)
response = api.tables.apps_graphql_query(
    app=APP_ID, table=ORDERS_TABLE_NAME, graph_ql_request=request
)

if response.errors:
    print(f"Order fetch errors: {response.errors}")
else:
    records = response.data.get(f"app_{APP_ID}_{ORDERS_TABLE_NAME}") or []
    print(records[0] if records else "Order not found")
```

#### Example: Update `delivered` flag by `orderId`

```python
from typing import Any, Dict
from automio import GraphQLRequest
from automio.wrapper import API

api = API()
APP_ID = 130
ORDERS_TABLE_NAME = "orders"

mutation = f"""mutation UpdateOrderDelivered($order_id: String!, $delivered: Boolean!, $organization_id: Int) {{
    update_app_{APP_ID}_{ORDERS_TABLE_NAME}(
        where: {{orderId: {{_eq: $order_id}}, organization_id: {{_eq: $organization_id}}}},
        _set: {{delivered: $delivered}}
    ) {{
        affected_rows
    }}
}}"""

variables: Dict[str, Any] = {"order_id": "eba6bc90-faab-11f0-96f9-c518c5ba21fb", "delivered": True, "organization_id": 2}
request = GraphQLRequest(query=mutation, variables=variables)
response = api.tables.apps_graphql_query(
    app=APP_ID, table=ORDERS_TABLE_NAME, graph_ql_request=request
)

if response.errors:
    print(f"Order update errors: {response.errors}")
```

#### Example: Query without variables (simple string query)

```python
from automio import GraphQLRequest
from automio.wrapper import API

api = API()
APP_ID = 130
TABLE_NAME = "billing_entries"

# GraphQLRequest can be used without variables for simple queries
query = f"""{{
    app_{APP_ID}_{TABLE_NAME}(
        order_by: {{ occurredAt: desc }}
        limit: 1
    ) {{
        allegroId
    }}
}}"""

response = api.tables.apps_graphql_query(
    app=APP_ID, table=TABLE_NAME, graph_ql_request=GraphQLRequest(query=query)
)
rows = response.data.get(f"app_{APP_ID}_{TABLE_NAME}", [])
print(rows[0]["allegroId"] if rows else "No records")
```

#### Example: Batch upsert with `on_conflict`

```python
from typing import Any, Dict, List
from automio import GraphQLRequest
from automio.wrapper import API

api = API()
APP_ID = 130
TABLE_NAME = "billing_entries"

def upsert_batch(rows: List[Dict[str, Any]]) -> None:
    mutation = f"""mutation InsertBatch($objects: [app_{APP_ID}_{TABLE_NAME}_insert_input!]!) {{
        insert_app_{APP_ID}_{TABLE_NAME}(
            objects: $objects,
            on_conflict: {{
                constraint: uq_app_{APP_ID}_{TABLE_NAME}_allegroId_org,
                update_columns: [occurredAt, type, value, balance]
            }}
        ) {{
            affected_rows
        }}
    }}"""

    request = GraphQLRequest(query=mutation, variables={"objects": rows})
    response = api.tables.apps_graphql_query(
        app=APP_ID, table=TABLE_NAME, graph_ql_request=request
    )
    if response.errors:
        print(f"Batch upsert errors: {response.errors}")
        return
    affected = (
        response.data
        .get(f"insert_app_{APP_ID}_{TABLE_NAME}", {})
        .get("affected_rows", 0)
    )
    print(f"Saved {affected} rows")
```

#### Example: Insert a message linked to an order

```python
from typing import Any, Dict
from automio import GraphQLRequest
from automio.wrapper import API

api = API()
APP_ID = 130
MESSAGES_TABLE_NAME = "messages"

message_mutation = f"""mutation InsertMessage($data: app_{APP_ID}_{MESSAGES_TABLE_NAME}_insert_input!) {{
    insert_app_{APP_ID}_{MESSAGES_TABLE_NAME}_one(object: $data) {{
        order_id
    }}
}}"""

send_message_payload: Dict[str, Any] = {
    "order_id": "c9a1d25a-fcf3-4950-a660-2f5b04a25320",  # foreign key id
    "messageContent": "Hello, user",
    "organization_id": 2,
}

request = GraphQLRequest(query=message_mutation, variables={"data": send_message_payload})
response = api.tables.apps_graphql_query(
    app=APP_ID, table=MESSAGES_TABLE_NAME, graph_ql_request=request
)

if response.errors:
    print(f"Message insert errors: {response.errors}")
else:
    print(response.data)
```

### Integration API Calls

#### `api.integrations.call()`

Make synchronous API calls to integrated services.

```python
# Basic API call (default connection)
integration = api.integrations.get_integration("github")
data, response = api.integrations.call(
    integration,
    method="GET",
    endpoint="/user"
)

# With parameters and specific connection (connection_id is an integer ID)
integration = api.integrations.get_integration("allegro_sandbox")
for connection_id in integration.connected_accounts:
    data, response = api.integrations.call(
        integration,
        method="GET",
        endpoint="/sale/offers",
        params={"limit": 10, "offset": 0},
        connection_id=connection_id
    )

print(integration.user_connections)  # {4: "foobar@baz.com"}

# POST with JSON payload
data, response = api.integrations.call(
    integration,
    method="POST",
    endpoint="/messages/submit",
    json={
        "from": {"name": "John Doe", "address": "john@example.com"},
        "to": [{"name": "Jane Doe", "address": "jane@example.com"}],
        "subject": "Test Email",
        "text": "Hello from Python client!"
    },
    connection_id=connection_id
)

# With custom request headers
data, response = api.integrations.call(
    integration,
    method="GET",
    endpoint="/sale/offers",
    headers={"Accept": "application/vnd.allegro.public.v1+json", "X-Custom-Header": "value"},
    connection_id=connection_id
)

# Raw binary body (e.g. image upload via PUT)
declaration, _ = api.integrations.call(
    integration,
    method="POST",
    endpoint="/messaging/message-attachments",
    data={"filename": "photo.jpg", "size": len(image_bytes)},
    headers={"Content-Type": "application/vnd.allegro.public.v1+json"},
    connection_id=connection_id,
)
upload_data, upload_response = api.integrations.call(
    integration,
    method="PUT",
    endpoint=f"/messaging/message-attachments/{declaration['id']}",
    raw_body=image_bytes,
    headers={"Content-Type": "image/jpeg"},
    connection_id=connection_id,
)
```

Notes:

- `connection_id` is the integration connection ID.
- `organization_id` is optional.
- `headers` (optional) — custom HTTP headers forwarded to the external API. The integration's authorization headers are always added automatically by the backend; use `headers` only for extra headers required by the API (e.g. `Accept`, versioning headers).
- `raw_body` (optional) — raw `bytes` to send as the request body. Use this when the external API expects binary content (images, PDFs, etc.) instead of JSON or multipart. The `Content-Type` from `headers` is forwarded as the body content type.
- In `app_action`, when `connection_id` comes from `APP_CONNECTIONS_BY_INTEGRATION`, the backend can infer the target organization automatically. You do not need to pass `organization_id`.

> **Note:** `connection_id` is an **integer** ID. Use `integration.connected_accounts` to iterate over all connected account IDs, `integration.user_connections` to get `{connection_id: name}`, or `api.integrations.get_table_connections()` to get full table-scoped connection objects.

**Returns:** `(data, response)` tuple where:

- `data`: Parsed JSON response data
- `response`: HTTP response object with `.ok`, `.status_code`, `.text`, `.headers`, `.url` attributes

**Parallel Methods Return:** `List[ParallelCallResponse]` objects where each response contains:

- `.api_data`: Parsed JSON response data from the external service
- `.api_response`: HTTP response object (same as above)
- `.connection_id`: ID of the connection used for this request
- `.organization_id`: Organization ID associated with the connection

### Parallel API Calls

#### `api.integrations.parallel_call()`

Execute multiple API calls concurrently using different connections per request for maximum performance and flexibility.

```python
# Multi-connection parallel calls (recommended for scaling across accounts)
responses = api.integrations.parallel_call(
    integration,
    method="GET",
    endpoint="/sale/product-offers/{offerId}",
    data_list=[
        {
            "connection_id": 123,
            "organization_id": 456,
            "url_params": {"offerId": "offer_1"},
            "params": {"include": "details"}
        },
        {
            "connection_id": 789,
            "organization_id": 101,
            "url_params": {"offerId": "offer_2"},
            "data": {"expand": True}
        }
    ],
    table="app_123_products"
)

# Process results with new object-based API
for response in responses:
    print(f"Connection {response.connection_id}, Org {response.organization_id}")
    if response.api_response.ok:
        print(f"Offer: {response.api_data['name']}")
        print(f"Status: {response.api_response.status_code}")
    else:
        print(f"Error: {response.api_response.text}")
```

#### `api.integrations.parallel_call_for_connection()`

Execute multiple API calls in parallel using a single connection (simpler for single-account operations).

```python
# Single-connection parallel calls
responses = api.integrations.parallel_call_for_connection(
    integration,
    method="GET",
    endpoint="/sale/product-offers/{offerId}",
    data_list=[
        {"url_params": {"offerId": "123"}},
        {"url_params": {"offerId": "456"}},
        {"url_params": {"offerId": "789"}}
    ],
    connection_id="user@domain.com",
    organization_id=123,
    table="app_124_offers"
)

# Process results
for response in responses:
    if response.api_response.ok:
        print(f"Offer: {response.api_data['name']}")
    else:
        print(f"Error: {response.api_response.text}")
```

#### `api.integrations.parallel_call_stream()`

Stream parallel API calls for real-time processing and progress visibility.

```python
# Stream parallel calls with real-time processing (single connection)
for result in api.integrations.parallel_call_stream(
    integration,
    method="GET",
    endpoint="/sale/product-offers/{offerId}",
    data_list=[{"url_params": {"offerId": offer["id"]}} for offer in offers],
    connection_id=connection_id
):

    if result.get("final"):
        print(f" Completed! Processed {result['total_processed']} requests")
        break
    else:
        # Process individual response as it arrives
        data = result["data"]
        response = result["response"]
        index = result["index"]

        if response.ok:
            # Process immediately without waiting for all requests
            processed_data = process_data(data)
            save_to_database(processed_data)
        else:
            print(f"âťŚ Error for request {index}: {response.text}")
```

### Plugins

Plugins let you run shared, reusable Python code per integration - defined once in a Plugin Interface, implemented separately for each integration.

**Concepts:**

- **Plugin Interface** - a contract identified by a `namespace` and a list of `capabilities` (function names). Think of it as an abstract class. Example: `test_interface` with capability `foo`.
- **Plugin Interface code** - shared Python code available to all plugins under that interface (helpers, base functions, common constants).
- **Plugin** - a concrete implementation of that interface for a specific integration (e.g., `allegro_sandbox`). It can override selected capabilities and reuse the interface code.

#### Shared Interface Code

When a `Plugin Interface` defines code, plugins in that interface can import it with:

```python
from plugins.interfaces.<namespace> import <name>
```

Example for namespace `test_interface`:

```python
from plugins.interfaces.test_interface import test_ifc
```

Or import everything from the interface module:

```python
from plugins.interfaces.test_interface import *
```

#### `api.plugins.get_list(namespace)`

List all enabled plugins registered under a given interface namespace.

```python
for plugin in api.plugins.get_list("test_interface"):
    print(f"Plugin ID: {plugin.id}")
    print(f"Connection IDs: {plugin.connection_ids}")
```

#### `api.plugins.invoke(plugin, integration, connection_ids, **payload)`

Invoke a capability from a plugin. The first argument is `"namespace.capability"`, the second is the integration ID string.

```python
integration = api.integrations.get_integration("allegro_sandbox")

for plugin in api.plugins.get_list("test_interface"):
    result = api.plugins.invoke(
        "test_interface.foo",        # namespace.capability
        integration.integration_id,  # integration name as string
        connection_ids=plugin.connection_ids,
    )
```

The plugin code (written in the platform's code editor) receives `connection_ids` as a keyword argument:

```python
# Plugin code for test_interface / allegro_sandbox integration:
def foo(connection_ids: list):
    for connection_id in connection_ids:
        data, response = api.integrations.call(
            api.integrations.get_integration("allegro_sandbox"),
            method="GET",
            endpoint="/sale/offers",
            connection_id=connection_id,
        )
        print(data)
```

#### `APP_CONNECTIONS_BY_INTEGRATION` in `app_action`

For actions of type `app_action`, the runtime injects:

```python
APP_CONNECTIONS_BY_INTEGRATION = {
    "allegro_sandbox": [
        {"organization_id": 3, "connection_ids": [34, 9438]},
        {"organization_id": 2, "connection_ids": [102]},
    ]
}
```

Use this variable only in `app_action`. It is not available in `user_action`, `table_action`, `trigger_action`, or `webhook_action`.

Typical usage:

```python
from automio.wrapper import API

api = API()

for integration_id, organization_groups in APP_CONNECTIONS_BY_INTEGRATION.items():
    for group in organization_groups:
        organization_id = group["organization_id"]
        connection_ids = group["connection_ids"]

        result = api.plugins.invoke(
            "test_interface.foo",
            integration_id,
            connection_ids=connection_ids,
        )
        print(f"Processed org {organization_id}: {result}")
```

You can also call integration APIs directly from `app_action` using only `connection_id`:

```python
for group in APP_CONNECTIONS_BY_INTEGRATION["allegro_sandbox"]:
    for connection_id in group["connection_ids"]:
        data, response = api.integrations.call(
            api.integrations.get_integration("allegro_sandbox"),
            method="GET",
            endpoint="/sale/offers",
            connection_id=connection_id,
        )
        print(group["organization_id"], connection_id, data)
```

Notes:

- Keys are `integration_id` values, for example `allegro_sandbox` or `gmail`.
- Each integration contains a list of organization groups.
- Each organization group has:
  - `organization_id`: the organization that granted the connections to the application,
  - `connection_ids`: the enabled connection IDs for that integration and organization.
- The data includes only connections explicitly enabled for the app through `ApplicationIntegrationAccountAccess`.
- The delegated `api.integrations.call(..., connection_id=...)` flow described above is available only in `app_action`.

### Connection Management

#### `integration.connected_accounts`

The simplest way to iterate over all connected account IDs for an integration. Returns an iterable of integer connection IDs.

```python
integration = api.integrations.get_integration("allegro_sandbox")
for connection_id in integration.connected_accounts:
    data, response = api.integrations.call(
        integration,
        method="GET",
        endpoint="/billing/billing-entries",
        connection_id=connection_id,
        params={"limit": 100},
    )
```

#### `integration.user_connections`

Contains the same connections as `integration.connected_accounts`, but as a dictionary where each key is the connection ID and each value is the display name.

```python
integration = api.integrations.get_integration("allegro_sandbox")

for connection_id, name in integration.user_connections.items():
    print(connection_id, name)
    data, response = api.integrations.call(
        integration,
        method="GET",
        endpoint="/billing/billing-entries",
        connection_id=connection_id,
    )
```

#### `api.integrations.get_table_connections()`

Get all connected accounts for an integration. Returns list of connections with `id` and `name`.

```python
# Get all connected accounts
connections = api.integrations.get_table_connections(integration: Integration, table_name: str)
for connection in connections:
    print(f"ID: {connection.id}, Name: {connection.name}")

    # Use connection ID in API calls
    data, response = api.integrations.call(
        integration,
        method="GET",
        endpoint="/sale/offers",
        connection_id=connection.id  # integer ID
    )
```

### Caching System

The SDK exposes two cache scopes:

| API | Scope | Typical use |
|---|---|---|
| `api.app_cache` | current app + current organization | data reused by actions in the same app for the same organization |
| `api.org_cache` | current organization | data shared between apps and actions in the same organization |

`api.user_cache` is kept as a backwards-compatible alias for `api.org_cache`, but new code should use `api.org_cache`.

Both cache APIs expose:

- `set(key, value, ttl=None)`
- `get(key)`
- `delete(key)`

#### Response Shape

`get()` returns cache metadata, not only the raw value:

```python
{
    "key": "api_data",
    "value": {"rates": [1.2, 1.5, 1.8]},
    "exists": True,
    "ttl": 300,
}
```

For a missing key:

```python
{
    "key": "api_data",
    "value": None,
    "exists": False,
    "ttl": None,
}
```

Always read your cached data from the `value` field after checking `exists`.

#### App Cache

Shared cache for the current app and organization. Use it for data that should not cross app boundaries.

```python
api.app_cache.set(
    key="api_data",
    value={"rates": [1.2, 1.5, 1.8]},
    ttl=300  # 5 minutes
)

cached_data = api.app_cache.get(key="api_data")
if cached_data and cached_data["exists"]:
    print(cached_data["value"])  # {'rates': [1.2, 1.5, 1.8]}

# Delete cache entry
api.app_cache.delete(key="api_data")
```

#### Organization Cache

Shared cache for the current organization, independent of the app. Use it when multiple apps or actions in the same organization should share the same cached value.

```python
api.org_cache.set(
    key="my_key",
    value="my_value",
    ttl=3600  # 1 hour
)

cached = api.org_cache.get(key="my_key")
if cached and cached["exists"]:
    print(cached["value"])

# Delete organization cache
api.org_cache.delete(key="my_key")
```

#### TTL

`ttl` is optional and is expressed in seconds.

```python
api.org_cache.set("temporary_state", {"step": 2}, ttl=900)
```

If `ttl` is omitted, the value remains until deleted or evicted by the underlying Redis/cache service.

#### Cache Keys

Keys are strings. The SDK supports keys with slashes and spaces, but stable simple names are easier to inspect and clean up.

```python
api.app_cache.set("imports/allegro/page-1", {"processed": 100}, ttl=900)
```

#### Errors and Availability

If the hosted cache service returns an error response, cache calls raise `CacheServiceError`.

If cache cannot be initialized in the current environment, `api.app_cache`, `api.org_cache`, `api.user_cache`, and `api.task` still exist, but using them raises `CacheUnavailableError` with the initialization reason.

### Notifications

#### `api.runtime.send_notification()`

Send an in-app notification to the current runtime user. The backend resolves
the recipient from the current `APP_TOKEN`/session, so do not pass a user id.

```python
api.runtime.send_notification(
    title="Import finished",
    description="128 rows are ready to review.",
    kind="success",  # info | success | warning | error
    event_key="import_finished",
    image="https://example.com/import-summary.png",
    data={"batch_id": "batch_123"},
    actions=[
        {
            "title": "Review rows",
            "url": "/my-app/imports/batch_123",
        },
    ],
)
```

`APP_ID` and the notification source are resolved automatically from the runtime
environment and headers. The backend stores runtime events under the
`app_runtime.` namespace automatically. `image` and `actions[].url` accept
`https://...` URLs or internal paths starting with `/`.

This method requires a valid hosted runtime `APP_TOKEN` for the current
`APP_ID`. Local direct-runs without that token raise a clear runtime error
instead of sending a notification.

#### `api.notify_me()`

Send notifications (via email) to yourself (useful for monitoring and alerts).

```python
# Send notification
api.notify_me(
    subject="API Process Completed",
    text="Successfully processed 1000 records",
    html="<h1>Success!</h1><p>Processed <strong>1000</strong> records</p>"
)
```

`notify_me` is an email helper with signature
`api.notify_me(subject, text, html=None)`. It is separate from in-app runtime
notifications.

### Task Progress Tracking

#### `api.task.set_progress(progress: int, info: str, status: 'pending' | 'completed' | 'failed')`

Update task progress for long-running operations.
`progress` and `info` are displayed in the UI during task execution, while `status` indicates the task state.

Allowed statuses:

| Status | Meaning |
|---|---|
| `pending` | Work is still running |
| `completed` | Work finished successfully |
| `failed` | Work failed |

```python
# Initialize progress
api.task.set_progress(0, "Starting data processing...", "pending")

# Update progress throughout your task
for i, item in enumerate(large_dataset):
    # Process item
    process_item(item)

    # Update progress every 100 items
    if i % 100 == 0:
        progress = int((i / len(large_dataset)) * 100)
        api.task.set_progress(
            progress,
            f"Processed {i}/{len(large_dataset)} items",
            "pending"
        )

# Complete the task
api.task.set_progress(100, "Processing completed!", "completed")
```

In the hosted cache service, `status="completed"` or `status="failed"` forces progress to `100`. A progress value of `100` with any other status is normalized to `completed`.

For local Redis development, progress is stored under `task:progress:{TASK_ID}` and logged to the console. In Automio runtime, progress updates are written to the code runner task state and displayed in the UI.

## Configuration & Environment

### Cache & Progress Service Configuration

The client automatically detects and adapts to your environment:

#### **Automio Runtime (Automatic)**

When running inside the Automio action runtime:

- Uses the hosted cache service API
- `TASK_ID` is automatically assigned by the platform
- `CACHE_SERVICE_URL` and `CACHE_SERVICE_TOKEN` are provided by the platform
- Progress tracking is displayed in the Automio UI
- `app_cache` is scoped to the current app and organization
- `org_cache` is scoped to the current organization and can be shared between apps
- The cache service derives the real namespace from `TASK_ID`, so action code does not pass app or organization IDs manually

#### **Local Development (Redis)**

When running locally without `CACHE_SERVICE_URL`:

- Automatically uses direct Redis connection
- `TASK_ID` is generated as a random UUID4
- Progress is logged to console
- Requires Redis server running locally
- Uses `ORGANIZATION_ID` or `ACTION_OWNER_ORGANIZATION_ID` for local org cache namespacing; falls back to `local`

Local Redis namespaces match hosted runtime semantics:

```text
app:{APP_ID}:code_owner:{ORGANIZATION_ID}:{key}
org:{ORGANIZATION_ID}:{key}
```

#### Optional Local Redis Dependency

The hosted Automio runtime does not require the Redis Python package in your action dependencies. Install `redis` yourself only when you want to use the direct local Redis fallback without `CACHE_SERVICE_URL`.

```bash
pip install redis
```

### Usage Impact

**From the user perspective, there are no code changes required.** The client automatically:

- **Detects environment** and chooses appropriate backend
- **Maintains consistent API** across both configurations
- **Handles serialization/deserialization** transparently
- **Provides same response format** regardless of backend

```python
# This code works identically in both environments
api.app_cache.set("my_key", {"data": "value"}, ttl=300)
cached_data = api.app_cache.get("my_key")
if cached_data and cached_data["exists"]:
    print(cached_data["value"])
api.task.set_progress(50, "Halfway done", "pending")
```

### Cache Use-Case Script

The repository includes a plain Python script that exercises common cache scenarios using the current `.env`:

```bash
python cache_usecase.py
```

It checks:

- `api.app_cache` and `api.org_cache`
- JSON and scalar roundtrips
- missing keys
- overwrites
- deletion
- TTL expiry
- keys with slashes/spaces
- task progress
- optional `api.user_cache` alias with `--include-user-cache-alias`

Useful flags:

```bash
python cache_usecase.py --skip-ttl
python cache_usecase.py --skip-progress
python cache_usecase.py --include-user-cache-alias
```

#### Environment Detection Logic

```python
# The client automatically determines configuration:
if CACHE_SERVICE_URL:
    # Use Automio cache service API
    # - HTTP requests to cache service
    # - Progress updates sent to platform
    # - Task ID from runtime environment
else:
    # Use local Redis connection
    # - Direct Redis operations
    # - Progress logged to console
    # - UUID4 generated for task tracking
```

### Integration Management

#### `api.integrations.get_integration()`

Get integration configuration for API calls.

```python
# Get integration
integration = api.integrations.get_integration("allegro_sandbox")
# Now use this integration object in .call() methods

print(integration.user_connections)  # {4: "foobar@baz.com"}
```

#### `api.integrations.get_integrations()`

List all available integrations.

```python
integrations = api.integrations.get_integrations()
for integration in integrations.results:
    print(f"ID: {integration.integration_id}, Name: {integration.name}")
```

## Common Patterns

### 1. **Parallel Data Processing**

```python
# Efficient parallel processing pattern
integration = api.integrations.get_integration("api_service")

# Step 1: Get list of items
list_data, response = api.integrations.call(
    integration, method="GET", endpoint="/items",
    params={"limit": 50}, connection_id="user@domain.com"
)

# Step 2: Process details in parallel (single connection)
detail_requests = [
    {"url_params": {"id": item["id"]}}
    for item in list_data["items"]
]

responses = api.integrations.parallel_call_for_connection(
    integration,
    method="GET",
    endpoint="/items/{id}/details",
    data_list=detail_requests,
    connection_id="user@domain.com",
    organization_id=123
)

# Step 3: Process results with new object API
processed_data = []
for response in responses:
    if response.api_response.ok:
        processed_data.append(transform_data(response.api_data))
    else:
        print(f"Error for connection {response.connection_id}: {response.api_response.text}")
```

### 2. **Multi-Connection Parallel Processing**

```python
# Advanced: Parallel processing across multiple connections simultaneously
integration = api.integrations.get_integration("api_service")

# Get all connections for the service
table_connections = api.integrations.get_table_connections(integration, "data_table")

# Build parallel requests for multiple connections
data_list = []
for org_id, connections in table_connections.items():
    for connection in connections:
        data_list.append({
            "connection_id": connection["id"],
            "organization_id": org_id,
            "params": {"limit": 50},
            "url_params": {"account_id": connection["id"]}
        })

# Execute all requests in parallel across all connections
responses = api.integrations.parallel_call(
    integration,
    method="GET",
    endpoint="/accounts/{account_id}/data",
    data_list=data_list,
    table="app_123_consolidated_data"
)

# Process results with full traceability
for response in responses:
    print(f"Org {response.organization_id}, Connection {response.connection_id}")
    if response.api_response.ok:
        save_data(response.api_data, response.connection_id, response.organization_id)
    else:
        log_error(response.connection_id, response.api_response.text)
```

### 3. **Multi-Account Operations**

```python
# Process data across multiple connected accounts
connections = api.integrations.get_table_connections("service_name", "table_name")

for connection in connections:
    print(f"Processing account: {connection.name}")

    data, response = api.integrations.call(
        integration,
        method="GET",
        endpoint="/data",
        connection_id=connection.id
    )

    if response.ok:
        # Process account-specific data
        process_account_data(data, connection.id)
```

### 3. **Progress Tracking with Caching**

```python
def long_running_task():
    api.task.set_progress(0, "Initializing...", "in_progress")

    # Cache intermediate results
    api.app_cache.set("task_checkpoint", {"processed": 0}, ttl=3600)

    for i in range(1000):
        # Do work
        process_item(i)

        # Update progress and cache
        if i % 100 == 0:
            progress = int((i / 1000) * 100)
            api.task.set_progress(progress, f"Processed {i}/1000", "in_progress")
            api.app_cache.set("task_checkpoint", {"processed": i}, ttl=3600)

    api.task.set_progress(100, "Completed!", "completed")
    api.app_cache.delete("task_checkpoint")
```

## Performance Tips

- **Use parallel calls** for multiple API requests to the same service
- **Cache frequently accessed data** to reduce API calls
- **Stream parallel calls** for real-time processing of large datasets
- **Set appropriate TTL** for cached data based on update frequency
- **Monitor progress** for long-running tasks to improve user experience

## Error Handling

### Single API Calls

```python
data, response = api.integrations.call(integration, method="GET", endpoint="/data")

if not response.ok:
    print(f"API Error: {response.status_code} - {response.text}")
    return

if not data:
    print("No data received")
    return

# Process successful response
process_data(data)
```

### Parallel API Calls

```python
responses = api.integrations.parallel_call(
    integration, method="GET", endpoint="/data",
    data_list=requests_data
)

for response in responses:
    if not response.api_response.ok:
        print(f"Error for connection {response.connection_id}: {response.api_response.text}")
        continue

    if not response.api_data:
        print(f"No data received for connection {response.connection_id}")
        continue

    # Process successful response
    process_data(response.api_data, response.connection_id)
```
