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":
                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:
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 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},
    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

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.