> ## Documentation Index > Fetch the complete documentation index at: https://docs.bigdata.com/llms.txt > Use this file to discover all available pages before exploring further. # Workflows API Quickstart Guide > Get started with the Bigdata Workflows API for templated, reproducible research # 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](https://platform.bigdata.com/api-keys) ## Execute with Inline Template The simplest way to execute a workflow is with an inline template definition: ```python theme={null} 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: ```python theme={null} 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 ```python theme={null} 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 ```python theme={null} 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 ```python theme={null} # 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. Reference for every message type, including the difference between the Workflows `delta` envelope and the Research Agent `message` envelope. Turn `GROUNDING` references into inline Markdown citations. HTTP-level errors and in-stream `ERROR` / `TOOL_ERROR` / `LLM_RETRY` events. Resume a workflow execution by passing `execution_id` on the next request. Template anatomy, input placeholders, content filters, and research plans. How `research_plan` controls dynamic vs. structured execution. Discover and customize shared templates. Ready-to-use financial analysis templates.