Why Use the Workflows API?

The Workflows API enables templated, reproducible research designed for automation and production use cases. Unlike the conversational Research Agent, Workflows uses parameterized templates that produce consistent outputs.

Templated Research: Define prompts with Jinja2 placeholders for consistent execution
Reproducible Results: Same template + inputs = consistent outputs
Production Ready: Designed for batch processing and scheduled reports
Template Management: Create, store, clone, and share research templates

In this quickstart, you’ll learn how to:

Create your API Key
Execute a workflow with an inline template
Execute a workflow with a stored template ID
Process streaming responses
Manage templates (create, list, clone)

Create your API Key

Please click here to create your API Key: Developer Platform > API Keys

Execute with Inline Template

The simplest way to execute a workflow is with an inline template definition:

import requests
import json
from dotenv import load_dotenv
import os

# Load environment variables
load_dotenv()
my_api_key = os.getenv('BIGDATA_API_KEY')

endpoint = "https://agents.bigdata.com/v1/workflow/execute"

# Define an inline template with a parameterized prompt
payload = {
    "template": {
        "name": "Company Analysis",
        "prompt": "Analyze the financial health and recent developments for {{ company_id }}. Include key metrics, recent news, and investment outlook.",
        "expected_input": {
            "company_id": {"type": "rp_entity_id"}
        }
    },
    "input": {
        "company_id": "D8442A"  # NVIDIA's entity ID
    },
    "time_range": "last_30_days",
    "model_name": "base"
}

headers = {
    "X-API-KEY": my_api_key,
    "Content-Type": "application/json"
}

print("Starting workflow execution...")

with requests.post(endpoint, headers=headers, json=payload, stream=True, timeout=120) as r:
    r.raise_for_status()
    for raw_line in r.iter_lines(decode_unicode=True):
        if not raw_line or not raw_line.startswith("data: "):
            continue

        data = raw_line[6:].strip()
        try:
            event = json.loads(data)
            delta = event.get("delta", {})
            message_type = delta.get("type")

            if message_type == "ANSWER":
                content = delta.get("content", "")
                print(content, end="", flush=True)

            elif message_type == "COMPLETE":
                consumption = delta.get("consumption", [])
                print("\n\nWorkflow complete!")
                for c in consumption:
                    if c["type"] in ("base", "pro"):
                        print(f"  {c['type']}: {c['input_tokens']} input, {c['output_tokens']} output tokens")

        except json.JSONDecodeError:
            pass

Execute with Template ID

For production use, store templates and reference them by ID:

import requests
import json
from dotenv import load_dotenv
import os

load_dotenv()
my_api_key = os.getenv('BIGDATA_API_KEY')

# First, create and save a template
create_endpoint = "https://agents.bigdata.com/v1/workflow/templates"

template_def = {
    "name": "Credit Issuer Analysis",
    "description": "Comprehensive credit analysis for bond issuers",
    "prompt": """Analyze {{ company_id }} from a credit perspective. Include:
1. Business overview and credit profile
2. Recent financial performance and leverage metrics
3. Key credit risks and mitigants
4. Debt maturity profile and liquidity position
5. Rating agency outlook and recent actions
6. Investment recommendation with credit spread view""",
    "expected_input": {
        "company_id": {"type": "rp_entity_id"}
    }
}

headers = {
    "X-API-KEY": my_api_key,
    "Content-Type": "application/json"
}

# Create the template
response = requests.post(create_endpoint, headers=headers, json=template_def)
template = response.json()
template_id = template["id"]
print(f"Created template: {template_id}")

# Now execute using the template ID
execute_endpoint = "https://agents.bigdata.com/v1/workflow/execute"

payload = {
    "template": template_id,  # Reference by ID
    "input": {
        "company_id": "4E4980"  # Apple's entity ID
    },
    "time_range": "last_90_days"
}

print("\nExecuting workflow...")

with requests.post(execute_endpoint, headers=headers, json=payload, stream=True, timeout=120) as r:
    r.raise_for_status()
    for raw_line in r.iter_lines(decode_unicode=True):
        if not raw_line or not raw_line.startswith("data: "):
            continue

        data = raw_line[6:].strip()
        try:
            event = json.loads(data)
            delta = event.get("delta", {})
            message_type = delta.get("type")

            if message_type == "ANSWER":
                print(delta.get("content", ""), end="", flush=True)
            elif message_type == "COMPLETE":
                print("\n\nWorkflow complete!")

        except json.JSONDecodeError:
            pass

Understanding the Streaming Response

The Workflows API streams responses using Server-Sent Events. Key message types:

Type	Description
`THINKING`	Agent’s internal reasoning
`PLANNING`	Research plan with steps and status
`ACTION`	Tool calls (search queries, etc.)
`ANSWER`	The actual research output
`GROUNDING`	Source references with character offsets
`AUDIT`	Detailed search results
`COMPLETE`	Final message with resource consumption

Processing Planning Messages

if message_type == "PLANNING":
    plan = delta.get("plan", {})
    print(f"Plan: {plan.get('title')}")
    for step in plan.get("steps", []):
        status_icon = {
            "NOT_STARTED": "[ ]",
            "IN_PROGRESS": "[>]",
            "COMPLETED": "[x]",
            "SKIPPED": "[-]",
            "FAILED": "[!]"
        }.get(step.get("status"), "[ ]")
        print(f"  {status_icon} {step.get('description')}")

Template Management

List Your Templates

response = requests.get(
    "https://agents.bigdata.com/v1/workflow/templates",
    headers={"X-API-KEY": my_api_key}
)
templates = response.json()

for t in templates["templates"]:
    print(f"- {t['id']}: {t['name']}")

Clone a Community Template

# List community templates
response = requests.get(
    "https://agents.bigdata.com/v1/workflow/templates/community",
    headers={"X-API-KEY": my_api_key}
)
community = response.json()

# Clone the first one
template_id = community["templates"][0]["id"]
clone_response = requests.post(
    f"https://agents.bigdata.com/v1/workflow/templates/{template_id}/clone",
    headers={"X-API-KEY": my_api_key}
)
cloned = clone_response.json()
print(f"Cloned template: {cloned['id']}")

Next Steps

The example above only handles ANSWER and COMPLETE events. The Concepts guides explain every message type the stream can emit, how to render citations from GROUNDING references, how to handle errors robustly, and how to continue a workflow execution across multiple turns.

Streaming responses

Reference for every message type, including the difference between the Workflows delta envelope and the Research Agent message envelope.

Grounding and citations

Turn GROUNDING references into inline Markdown citations.

Error handling

HTTP-level errors and in-stream ERROR / TOOL_ERROR / LLM_RETRY events.

Conversation continuity

Resume a workflow execution by passing execution_id on the next request.

Creating Templates

Template anatomy, input placeholders, content filters, and research plans.

Research Plans

How research_plan controls dynamic vs. structured execution.

Community Templates

Discover and customize shared templates.

Example Templates

Ready-to-use financial analysis templates.

Getting Started

Research Service

Search Service

Upload proprietary content

Knowledge Graph

Watchlist

Troubleshooting

Workflows API Quickstart Guide

Why Use the Workflows API?

Create your API Key

Execute with Inline Template

Execute with Template ID

Understanding the Streaming Response

Processing Planning Messages

Template Management

List Your Templates

Clone a Community Template

Next Steps

Streaming responses

Grounding and citations

Error handling

Conversation continuity

Creating Templates

Research Plans

Community Templates

Example Templates

​Why Use the Workflows API?

​Create your API Key

​Execute with Inline Template

​Execute with Template ID

​Understanding the Streaming Response

​Processing Planning Messages

​Template Management

​List Your Templates

​Clone a Community Template

​Next Steps

Streaming responses

Grounding and citations

Error handling

Conversation continuity

Creating Templates

Research Plans

Community Templates

Example Templates

Why Use the Workflows API?

Create your API Key

Execute with Inline Template

Execute with Template ID

Understanding the Streaming Response

Processing Planning Messages

Template Management

List Your Templates

Clone a Community Template

Next Steps