# Botify Agent: Batch Run

**ID:** `batch_run`

Run any other tool over many records, from a list, a SQL query, a table or an uploaded CSV, and collect the results in a file or a table.

Run one tool over many records at once.

    This is the only way to fan a tool out: every other tool takes exactly one item.

    Sources: `rows` (records in the call itself), `sql` (a BigQuery query),
    `table` (a whole catalog table), `file` (a CSV/TSV uploaded at
    https://app.botify.com/:organization/:project/o/storage/tmp, or the output of
    a previous batch).

    Small sources run immediately and return their results. Larger ones run as a
    background job: you get a job_id to poll with batch_status, and the records
    never enter this conversation.

    The output is a temporary JSONL file whose URL is directly usable as the
    `file` source of a next batch, which is how several tools are chained. Each
    line is {row, item, response, error} -- the original row and mapped item are
    both kept, so results stay attributable and source columns survive chaining.

    Use tool_documentation on the target tool first: item_map has to match its
    item schema. Use dry_run to check a mapping against real records before
    paying for the whole source.
    

## Authentication

All endpoints require a **Bearer token** in the `Authorization` header.

```
Authorization: Bearer <your-token>
```

## Calling this agent

This agent supports the following processing modes:

| Mode | Type | Description |
|------|------|-------------|
| **Process** | Synchronous | Single-item processing. Best for real-time requests with immediate response. |
| **Batch Process** | Synchronous | Process multiple items in a single request for efficiency. |
| **Async Process** | Asynchronous | Single-item processing for long-running tasks that exceed timeout limits. |
| **Async Batch Process** | Asynchronous | Large-scale batch jobs with background processing. |

# Synchronous Processing

Synchronous calls block until the result is ready. Use these for quick operations where you need immediate results.

## Single Item Processing

Process a single input and receive the result immediately.

```
POST https://agents.botify.com/{org}/{project}/batch_run/process
```

### Request body

```json
{
  "item": {
    "tool": "<tool>",
    "source": "<source>"
  }
}
```

### Response

Returns a single processed item (HTTP 200).

### cURL example

```bash
curl -X POST "https://agents.botify.com/{org}/{project}/batch_run/process" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
  "item": {
    "tool": "<tool>",
    "source": "<source>"
  }
}'
```

## Batch Processing

Process multiple inputs in a single synchronous request. More efficient than making individual calls. Items are processed concurrently on the server.

```
POST https://agents.botify.com/{org}/{project}/batch_run/batch_process
```

### Request body

```json
{
  "items": [
    {
      "tool": "<tool>",
      "source": "<source>"
    }
  ]
}
```

### Response

Returns an array of results. Each element is either a successful processed item (`"status": "success"`) or an error (`"status": "error"`).

### cURL example

```bash
curl -X POST "https://agents.botify.com/{org}/{project}/batch_run/batch_process" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
  "items": [
    {
      "tool": "<tool>",
      "source": "<source>"
    }
  ]
}'
```

### When to use synchronous calls

- Processing completes within timeout limits (typically 30 seconds)
- You need immediate results for user-facing features
- Small to medium batch sizes (up to ~100 items)

# Asynchronous Processing

Asynchronous calls return immediately with a batch ID. You then poll for results using the `async_batches/` endpoints.

### Async processing flow

1. **Submit job** -- `POST` to `async_process` or `async_batch_process`
2. **Receive batch ID** -- response contains `batch_id`
3. **Poll status** -- `HEAD` request to check readiness (lightweight, no body)
4. **Get results** -- `GET` request to retrieve processed results

## Async Single Item

Submit a single item for background processing. Use the `/single` endpoint to retrieve the result.

```
POST https://agents.botify.com/{org}/{project}/batch_run/async_process
```

### Request body

```json
{
  "item": {
    "tool": "<tool>",
    "source": "<source>"
  }
}
```

### cURL example

```bash
# Step 1: Submit the job
curl -X POST "https://agents.botify.com/{org}/{project}/batch_run/async_process" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
  "item": {
    "tool": "<tool>",
    "source": "<source>"
  }
}'

# Response: { "batch_id": "abc123" }

# Step 2: Check if batch is ready (HEAD request -- lightweight check)
curl -I -X HEAD "https://agents.botify.com/{org}/{project}/batch_run/async_batches/abc123" \
  -H "Authorization: Bearer $TOKEN"
# Returns 200 if ready, 204 if still processing

# Step 3: Get the result (for single-item async)
curl -X GET "https://agents.botify.com/{org}/{project}/batch_run/async_batches/abc123/single" \
  -H "Authorization: Bearer $TOKEN"
# Returns 200 with result, 202 if still processing, 404 if batch not found
```

## Async Batch Processing

Submit multiple items for background processing. The request body uses **JSONL** (newline-delimited JSON) format.

```
POST https://agents.botify.com/{org}/{project}/batch_run/async_batch_process
```

### Request body (JSONL, `Content-Type: text/plain`)

The first line contains `config` and `batch_config`. Each subsequent line is an item with a unique `id`.

```
{"config": {}, "batch_config": {}}
{"id": "item_1", "item": {"tool": "<tool>", "source": "<source>"}}
```

### Response

Returns a JSON object with `batch_id`, `status`, and a `Location` header.

### cURL example

```bash
# Step 1: Submit the batch job
curl -X POST "https://agents.botify.com/{org}/{project}/batch_run/async_batch_process" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: text/plain" \
  -d '{"config": {}, "batch_config": {}}\n{"id": "item_1", "item": {"tool": "<tool>", "source": "<source>"}}'

# Response: { "batch_id": "xyz789" }

# Step 2: Check if batch is ready (HEAD request)
curl -I -X HEAD "https://agents.botify.com/{org}/{project}/batch_run/async_batches/xyz789" \
  -H "Authorization: Bearer $TOKEN"
# Returns 200 if ready, 204 if still processing

# Step 3: Get all results
curl -X GET "https://agents.botify.com/{org}/{project}/batch_run/async_batches/xyz789" \
  -H "Authorization: Bearer $TOKEN"
```

## Checking Batch Status

Use a lightweight `HEAD` request to check if your batch is ready without transferring data.

```
HEAD https://agents.botify.com/{org}/{project}/batch_run/async_batches/{batch_id}
```

```bash
curl -I -X HEAD "https://agents.botify.com/{org}/{project}/batch_run/async_batches/{batch_id}" \
  -H "Authorization: Bearer $TOKEN"

# Response headers indicate status:
# HTTP/1.1 200 OK        -> Batch is complete, results ready
# HTTP/1.1 204 No Content -> Still processing
# HTTP/1.1 404 Not Found -> Batch does not exist
```

## Retrieving Results

Once the batch is ready, fetch results using the appropriate endpoint.

```bash
# For async_process (single item) -- use /single endpoint
curl -X GET "https://agents.botify.com/{org}/{project}/batch_run/async_batches/{batch_id}/single" \
  -H "Authorization: Bearer $TOKEN"

# For async_batch_process (multiple items) -- use base endpoint
curl -X GET "https://agents.botify.com/{org}/{project}/batch_run/async_batches/{batch_id}" \
  -H "Authorization: Bearer $TOKEN"
# Returns streaming JSONL where each line is a processed item or error
```

### Async response codes

| Code | Endpoint | Meaning | Action |
|------|----------|---------|--------|
| 200 | `HEAD` / `GET` | Batch complete, results ready | Read results from response body |
| 204 | `HEAD` | Still processing | Continue polling |
| 202 | `GET /single` | Still processing | Continue polling |
| 400 | `GET /single` | All items failed (client error) | Check error in response body |
| 404 | `HEAD` / `GET` | Batch does not exist | Verify `batch_id` is correct |

> **HEAD vs GET:** Use `HEAD` requests for lightweight status checks (no response body). Use `GET` only when ready to retrieve results to minimize bandwidth.

### When to use asynchronous calls

- Processing takes longer than 30 seconds
- Large batch sizes (100+ items)
- Background processing where immediate results aren't required
- Integration with job queues or workflow systems

## Python Example

A complete Python example showing both synchronous and asynchronous patterns.

```python
import requests
import time

API_TOKEN = "your_api_token"
BASE_URL = "https://agents.botify.com/{organization}/{project}"
AGENT = "batch_run"

headers = {
    "Authorization": f"Bearer {API_TOKEN}",
    "Content-Type": "application/json"
}

single_payload = {
  "item": {
    "tool": "<tool>",
    "source": "<source>"
  }
}


def process_sync(payload: dict) -> dict:
    """Synchronous single-item processing."""
    response = requests.post(
        f"{BASE_URL}/{AGENT}/process",
        headers=headers,
        json=payload
    )
    response.raise_for_status()
    return response.json()


def process_async(
    payload: dict,
    poll_interval: int = 5,
    max_retries: int = 60
) -> dict:
    """Asynchronous single-item processing with polling."""
    # Submit job
    response = requests.post(
        f"{BASE_URL}/{AGENT}/async_process",
        headers=headers,
        json=payload
    )
    batch_id = response.json()["batch_id"]

    # Poll for results using HEAD (lightweight check)
    for _ in range(max_retries):
        status = requests.head(
            f"{BASE_URL}/{AGENT}/async_batches/{batch_id}",
            headers=headers
        )

        if status.status_code == 200:
            result = requests.get(
                f"{BASE_URL}/{AGENT}/async_batches/{batch_id}/single",
                headers=headers
            )
            return result.json()
        elif status.status_code == 204:
            time.sleep(poll_interval)
        else:
            raise Exception(f"Unexpected status: {status.status_code}")

    raise Exception("Max retries exceeded")


def process_async_batch(
    items: list,
    config: dict | None = None,
    poll_interval: int = 5
) -> dict:
    """Asynchronous batch processing with polling."""
    import json as _json

    # Build JSONL payload: first line is config, subsequent lines are items
    lines = [_json.dumps({"config": config or {}, "batch_config": {}})]
    for i, item in enumerate(items):
        lines.append(_json.dumps({"id": f"item_{i}", "item": item}))
    body = "\n".join(lines)

    # Submit batch (JSONL, text/plain)
    response = requests.post(
        f"{BASE_URL}/{AGENT}/async_batch_process",
        headers={**headers, "Content-Type": "text/plain"},
        data=body
    )
    batch_id = response.json()["batch_id"]

    # Poll until ready
    while True:
        status = requests.head(
            f"{BASE_URL}/{AGENT}/async_batches/{batch_id}",
            headers=headers
        )

        if status.status_code == 200:
            results = requests.get(
                f"{BASE_URL}/{AGENT}/async_batches/{batch_id}",
                headers=headers
            )
            return results.json()

        time.sleep(poll_interval)


# Example usage
# result = process_sync(single_payload)
# result = process_async(single_payload)
# results = process_async_batch([{"input": "item1"}, {"input": "item2"}])
```

## Billing

These usage SKUs can be charged on a call.

| SKU | Credits | Description |
| --- | ------- | ----------- |
| Tool call | 1 per request | Charged once per successful item. |


## Schemas

### Item

```json
{
  "type": "object",
  "properties": {
    "tool": {
      "description": "Id of the tool to run over every record. Use tool_documentation to read its item schema before writing item_map.",
      "title": "Tool",
      "type": "string"
    },
    "source": {
      "$ref": "#/$defs/SourceSpec",
      "description": "Where the records come from."
    },
    "item_map": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Projection from a source record onto the tool's item, using {{row.<column>}} (ex: {\"url\": \"{{row.page_url}}\"}). Omit it when the records already have the tool's item shape.",
      "title": "Item Map"
    },
    "sink": {
      "anyOf": [
        {
          "$ref": "#/$defs/SinkSpec"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Where results go. Defaults to a temporary gzipped JSONL file whose URL can be passed straight back as a file source."
    },
    "tool_config": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The target tool's own config, shared by every record.",
      "title": "Tool Config"
    },
    "dry_run": {
      "default": false,
      "description": "Run only the first 5 records and return them in full, launching nothing. Use it to check a mapping or a prompt against a large source in one turn.",
      "title": "Dry Run",
      "type": "boolean"
    }
  },
  "required": [
    "tool",
    "source"
  ],
  "$defs": {
    "FileFormat": {
      "enum": [
        "auto",
        "jsonl",
        "csv",
        "tsv"
      ],
      "type": "string"
    },
    "FileSink": {
      "description": "A temporary JSONL file in project storage, readable back as a source.",
      "properties": {
        "type": {
          "const": "file",
          "default": "file",
          "title": "Type",
          "type": "string"
        },
        "gzip": {
          "default": true,
          "description": "Compress the output.",
          "title": "Gzip",
          "type": "boolean"
        }
      },
      "title": "FileSink",
      "type": "object"
    },
    "FileSource": {
      "description": "Rows from an uploaded CSV/TSV, or from a previous batch's JSONL output.",
      "properties": {
        "type": {
          "const": "file",
          "default": "file",
          "title": "Type",
          "type": "string"
        },
        "url": {
          "description": "File URL. Upload a local file at https://app.botify.com/:organization/:project/o/storage/tmp and pass the returned URL, or reuse the sink URL of a previous batch. Public HTTP(S) URLs are accepted too. CSV, TSV and JSONL, gzip allowed.",
          "title": "Url",
          "type": "string"
        },
        "format": {
          "$ref": "#/$defs/FileFormat",
          "default": "auto",
          "description": "File format, or auto-detect."
        }
      },
      "required": [
        "url"
      ],
      "title": "FileSource",
      "type": "object"
    },
    "RowsSource": {
      "description": "Records passed directly in the call.",
      "properties": {
        "type": {
          "const": "rows",
          "default": "rows",
          "title": "Type",
          "type": "string"
        },
        "rows": {
          "description": "The records to process, one object per call.",
          "items": {
            "additionalProperties": true,
            "type": "object"
          },
          "minItems": 1,
          "title": "Rows",
          "type": "array"
        }
      },
      "required": [
        "rows"
      ],
      "title": "RowsSource",
      "type": "object"
    },
    "SinkSpec": {
      "discriminator": {
        "mapping": {
          "file": "#/$defs/FileSink",
          "table": "#/$defs/TableSink"
        },
        "propertyName": "type"
      },
      "oneOf": [
        {
          "$ref": "#/$defs/FileSink"
        },
        {
          "$ref": "#/$defs/TableSink"
        }
      ]
    },
    "SourceSpec": {
      "discriminator": {
        "mapping": {
          "file": "#/$defs/FileSource",
          "rows": "#/$defs/RowsSource",
          "sql": "#/$defs/SqlSource",
          "table": "#/$defs/TableSource"
        },
        "propertyName": "type"
      },
      "oneOf": [
        {
          "$ref": "#/$defs/RowsSource"
        },
        {
          "$ref": "#/$defs/SqlSource"
        },
        {
          "$ref": "#/$defs/TableSource"
        },
        {
          "$ref": "#/$defs/FileSource"
        }
      ]
    },
    "SqlSource": {
      "description": "Rows from a BigQuery query over the project's catalog.",
      "properties": {
        "type": {
          "const": "sql",
          "default": "sql",
          "title": "Type",
          "type": "string"
        },
        "query": {
          "description": "BigQuery Standard SQL over catalog tables. Prefix every table with \"catalog.\" (ex: catalog.crawl_pages_20251201). Same access as tables_query.",
          "title": "Query",
          "type": "string"
        },
        "limit": {
          "anyOf": [
            {
              "minimum": 1,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Stop after this many rows.",
          "title": "Limit"
        }
      },
      "required": [
        "query"
      ],
      "title": "SqlSource",
      "type": "object"
    },
    "TableSink": {
      "description": "A table in the project's editable catalog.",
      "properties": {
        "type": {
          "const": "table",
          "default": "table",
          "title": "Type",
          "type": "string"
        },
        "table": {
          "description": "Destination table in the catalog.",
          "pattern": "^[A-Za-z_][A-Za-z0-9_]{0,1023}$",
          "title": "Table",
          "type": "string"
        },
        "row_map": {
          "anyOf": [
            {
              "additionalProperties": true,
              "type": "object"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Column projection, using {{item.<field>}}, {{response.<field>}}, {{error}} and {{row.<column>}}. Defaults to `row`, `item`, `response` and `error` columns.",
          "title": "Row Map"
        },
        "replace_existing": {
          "default": false,
          "description": "Replace the table instead of appending to it.",
          "title": "Replace Existing",
          "type": "boolean"
        }
      },
      "required": [
        "table"
      ],
      "title": "TableSink",
      "type": "object"
    },
    "TableSource": {
      "description": "Every row of one catalog table.",
      "properties": {
        "type": {
          "const": "table",
          "default": "table",
          "title": "Type",
          "type": "string"
        },
        "table": {
          "description": "Table name, with or without the \"catalog.\" prefix.",
          "pattern": "^(?:catalog\\.)?[A-Za-z_][A-Za-z0-9_]*$",
          "title": "Table",
          "type": "string"
        },
        "limit": {
          "anyOf": [
            {
              "minimum": 1,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Stop after this many rows.",
          "title": "Limit"
        }
      },
      "required": [
        "table"
      ],
      "title": "TableSource",
      "type": "object"
    }
  }
}
```

### Config

```json
{
  "type": "object",
  "properties": {
    "on_error": {
      "$ref": "#/$defs/OnError",
      "default": "skip",
      "description": "skip: record the failure and carry on. fail_job: stop the whole batch on the first failing record."
    },
    "chunk_size": {
      "default": 50,
      "description": "Records sent to the tool at once.",
      "maximum": 500,
      "minimum": 1,
      "title": "Chunk Size",
      "type": "integer"
    },
    "inline_max_rows": {
      "default": 10,
      "description": "Above this many records, the batch runs as a job.",
      "maximum": 100,
      "minimum": 1,
      "title": "Inline Max Rows",
      "type": "integer"
    }
  },
  "required": [],
  "$defs": {
    "OnError": {
      "enum": [
        "skip",
        "fail_job"
      ],
      "type": "string"
    }
  }
}
```

### Response

```json
{
  "$defs": {
    "SinkPointer": {
      "description": "Where a batch's output landed.",
      "properties": {
        "type": {
          "enum": [
            "file",
            "table"
          ],
          "title": "Type",
          "type": "string"
        },
        "url": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Sink file URL, reusable as the source of a next batch.",
          "title": "Url"
        },
        "table": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Destination table.",
          "title": "Table"
        },
        "rows_written": {
          "default": 0,
          "description": "Records written.",
          "title": "Rows Written",
          "type": "integer"
        }
      },
      "required": [
        "type"
      ],
      "title": "SinkPointer",
      "type": "object"
    }
  },
  "description": "What a batch returns, whichever path it took.",
  "properties": {
    "mode": {
      "description": "dry_run and inline carry their results; job carries a job_id to poll with batch_status.",
      "enum": [
        "dry_run",
        "inline",
        "job"
      ],
      "title": "Mode",
      "type": "string"
    },
    "tool": {
      "description": "The tool that was run.",
      "title": "Tool",
      "type": "string"
    },
    "results": {
      "anyOf": [
        {
          "items": {
            "additionalProperties": true,
            "type": "object"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "One {item, response, error} object per record.",
      "title": "Results"
    },
    "job_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Poll it with batch_status.",
      "title": "Job Id"
    },
    "execution_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Underlying Cloud Run execution.",
      "title": "Execution Id"
    },
    "rows_total": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Records the batch will process.",
      "title": "Rows Total"
    },
    "rows_ok": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Records that succeeded.",
      "title": "Rows Ok"
    },
    "rows_error": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Records that failed.",
      "title": "Rows Error"
    },
    "estimated_credits": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Credits the batch is expected to cost.",
      "title": "Estimated Credits"
    },
    "sink": {
      "anyOf": [
        {
          "$ref": "#/$defs/SinkPointer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Where the output landed."
    },
    "message": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "One line about the outcome.",
      "title": "Message"
    }
  },
  "required": [
    "mode",
    "tool"
  ],
  "title": "BatchRunResult",
  "type": "object"
}
```

## Best Practices

**Use exponential backoff for polling.** Start with short intervals (1-2 seconds) and increase the delay between polls to reduce API load.

**Set reasonable timeouts.** For synchronous calls, configure your HTTP client with appropriate timeout values (30-60 seconds).

**Handle rate limits gracefully.** Implement retry logic with backoff when you receive 429 (Too Many Requests) responses.

**Batch when possible.** Use batch endpoints to reduce the number of API calls and improve throughput.