Skip to main content

​
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:
  1. Create your API Key
  2. Execute a workflow with an inline template
  3. Execute a workflow with a stored template ID
  4. Process streaming responses
  5. 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":
                usage = delta.get("usage", [])
                print("\n\nWorkflow complete!")
                for u in usage:
                    print(f"  {u['type']}: {u['input_tokens']} input, {u['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:
TypeDescription
THINKINGAgent’s internal reasoning
PLANNINGResearch plan with steps and status
ACTIONTool calls (search queries, etc.)
ANSWERThe actual research output
GROUNDINGSource references with character offsets
AUDITDetailed search results
COMPLETEFinal message with token usage

​
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},
    params={"limit": 20}
)
templates = response.json()

for t in templates["results"]:
    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["results"][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

Creating Templates

Learn template anatomy and best practices

Research Plans

Learn how research plans affect execution

Community Templates

Discover and customize shared templates

Example Templates

Ready-to-use financial analysis templates