Add Work Items via API

The Work Item API lets you programmatically send tasks to your Worker Agents. This allows you to trigger automated workflows from your applications, integrate with external systems, and build event-driven work.

Work Item API is only available in Sema4.ai Enterprise Edition on AWS.

Before You Start

You need three things:

Endpoint URL - Your organization's API endpoint from Control Room. Check this documentation on where to find your endpoint URL.
API Key - A bearer token for authentication (check this for how to get one)
Agent ID - The unique identifier of the Worker Agent you want to use. Follow the below steps to get your Agent ID.

Get your Agent ID

Log in to Control Room
Under the Help menu, select API Helper
Select List agents (GET) from the dropdown in Select API option
Choose your Workspace and API Key that you want to use
Execute to see a list of all agents in your organization
Find the Worker Agent you want to use and copy its id field

Create Your First Work Item

Let's create a Work Item for the Sales Order Processor we built in the previous tutorial.

curl --request POST \
  --url 'https://{endpoint_url}/api/v1/work-items' \
  --header 'Authorization: Bearer your_api_key_here' \
  --header 'Content-Type: application/json' \
  --data '{
    "agent_id": ""af57907d-646d-4a4e-9ced-9049af96xxxx", # replace with your Agent ID
    "payload": {
      "orderId": "12345",
      "customerName": "John Doe",
      "amount": 250.00,
      "productID": "ABCD",
      "quantity": 2
    },
    "messages": [
      {
        "role": "user",
        "content": [
          {
            "kind": "text",
            "text": "Process this new sales order from our online store"
          }
        ]
      }
    ]
  }'

This request will create a new Work Item and return a WorkItem object with details including the Work Item ID, status, and configuration.

What happens:

The Work Item is created with status PENDING
Your Worker Agent automatically picks it up
It extracts the order details, verifies the amount is between $0-$10,000
Logs the verification and generates a confirmation
Marks the Work Item as COMPLETED

That's it! You've just automated your first sales order processing.

Common Integration Scenarios

Worker Agents excel at processing tasks triggered by real-world events in your business systems. Here are the most common ways to integrate Work Items into your workflows. Instead of manually creating Work Items, you can automatically trigger them when specific events occur:

Email arrives - Process customer inquiries, extract data from email attachments, route support requests
File uploaded - Validate documents, extract information, trigger approval workflows
Scheduled time - Run daily reconciliations, generate reports, perform batch processing
Webhook received - React to external system events, process API callbacks, handle form submissions
Application event - Process user actions, handle payment confirmations, trigger notifications

Each scenario follows the same pattern: an event occurs → your code creates a Work Item → your Worker Agent processes it automatically.

These integration patterns can be combined and customized for your specific workflows. The key is to identify events in your business process that should trigger automated work, then let Worker Agents handle the processing reliably and at scale.

Trigger from Incoming Email

Use case: Automatically process customer orders, support requests, or document submissions that arrive via email.

How it works:

Microsoft Graph API monitors your mailbox for new emails
AWS Lambda function receives email notification
Lambda creates a Work Item with email content and attachments
Your Worker Agent processes the email according to your runbook

Setup requirements:

Microsoft 365 account with Graph API access
Azure AD app registration with Mail.Read permissions
AWS Lambda function to handle notifications
Configured webhook subscription in Microsoft Graph

Use AWS secrets manager or Azure Key Vault to securely store your Sema4.ai API key and other credentials.

Sample Implementation:

import requests
import json
 
def process_email_notification(email_data):
    """
    Process incoming email and create Work Item
    
    Args:
        email_data: Email notification from Microsoft Graph
    """
    # Extract email details
    subject = email_data.get('subject', '')
    sender = email_data.get('from', {}).get('emailAddress', {}).get('address', '')
    body = email_data.get('body', {}).get('content', '')
    has_attachments = email_data.get('hasAttachments', False)
    
    # Create Work Item
    response = requests.post(
        'https://{endpoint_url}/api/v1/work-items',
        headers={
            'Authorization': 'Bearer your_api_key_here',
            'Content-Type': 'application/json'
        },
        json={
            'agent_id': 'email_processor',
            'payload': {
                'subject': subject,
                'sender': sender,
                'has_attachments': has_attachments,
                'source': 'email'
            },
            'messages': [{
                'role': 'user',
                'content': [{
                    'kind': 'text',
                    'text': f'Process email from {sender}. Subject: {subject}. Extract key information and route accordingly.'
                }]
            }],
            'callbacks': [{
                'url': 'https://your-app.com/webhooks/email-processed',
                'on_status': 'COMPLETED'
            }]
        }
    )
    
    return response.json()

What the Worker Agent can do:

Extract order details from email body
Download and process attachments
Route to appropriate department
Send automated responses
Create tickets in your system

Complete setup guide: See the Microsoft Email Integration README for detailed instructions on setting up Microsoft Graph subscriptions and Lambda functions.

Trigger from S3 File Upload

Use case: Automatically process documents when they're uploaded to AWS S3 - invoices, contracts, receipts, CSV files, etc.

How it works:

File is uploaded to S3 bucket (manually or via another application)
S3 triggers Lambda function via event notification
Lambda downloads file and uploads to Work Item
Lambda finalizes Work Item with processing instructions
Worker Agent processes the document

Setup requirements:

AWS S3 bucket with event notifications enabled
AWS Lambda function with S3 read permissions
IAM role with appropriate permissions
Sema4.ai API credentials stored in AWS Secrets Manager

Implementation:

import json
import os
import boto3
import requests
import logging
from urllib.parse import unquote_plus
 
logger = logging.getLogger()
logger.setLevel(logging.INFO)
 
def lambda_handler(event, context):
    """
    Process S3 file upload and create Work Item
    """
    # Get configuration from environment variables
    workitems_api_url = os.environ['WORKITEMS_API_URL']
    api_key = os.environ['S4_API_KEY']
    agent_id = os.environ['AGENT_ID']
    webhook_url = os.environ.get('WEBHOOK_URL', '')
    
    s3_client = boto3.client('s3')
    
    # Process each S3 record
    for record in event.get('Records', []):
        # Extract S3 information
        bucket = record['s3']['bucket']['name']
        key = unquote_plus(record['s3']['object']['key'])
        
        logger.info(f'Processing file: {bucket}/{key}')
        
        # Download file from S3
        temp_file_path = f'/tmp/{os.path.basename(key)}'
        s3_client.download_file(bucket, key, temp_file_path)
        
        try:
            # Step 1: Upload file to create PRECREATED Work Item
            work_item_id = upload_file(
                api_url=workitems_api_url,
                api_key=api_key,
                file_path=temp_file_path,
                file_name=os.path.basename(key)
            )
            
            logger.info(f'Created Work Item: {work_item_id}')
            
            # Step 2: Finalize Work Item with processing instructions
            create_work_item(
                api_url=workitems_api_url,
                api_key=api_key,
                agent_id=agent_id,
                work_item_id=work_item_id,
                file_name=key,
                bucket=bucket,
                webhook_url=webhook_url
            )
            
            logger.info(f'Work Item {work_item_id} ready for processing')
            
        finally:
            # Clean up temporary file
            if os.path.exists(temp_file_path):
                os.remove(temp_file_path)
    
    return {
        'statusCode': 200,
        'body': json.dumps('Processing completed')
    }
 
def upload_file(api_url, api_key, file_path, file_name):
    """Upload file and get Work Item ID"""
    headers = {'Authorization': f'Bearer {api_key}'}
    
    with open(file_path, 'rb') as file:
        files = {'file': (file_name, file, 'application/octet-stream')}
        response = requests.post(
            f'{api_url}/work-items/upload-file',
            headers=headers,
            files=files,
            timeout=30
        )
        
        if response.status_code == 200:
            return response.json()['work_item_id']
        else:
            raise Exception(f'Upload failed: {response.text}')
 
def create_work_item(api_url, api_key, agent_id, work_item_id, file_name, bucket, webhook_url):
    """Finalize Work Item with agent assignment"""
    headers = {
        'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }
    
    payload = {
        'agent_id': agent_id,
        'work_item_id': work_item_id,
        'payload': {
            'file_name': file_name,
            'source_bucket': bucket,
            'upload_source': 's3'
        },
        'messages': [{
            'role': 'user',
            'content': [{
                'kind': 'text',
                'text': f'Process document {file_name} and extract key information'
            }]
        }],
        'callbacks': [
            {'url': webhook_url, 'on_status': 'COMPLETED'},
            {'url': webhook_url, 'on_status': 'NEEDS_REVIEW'}
        ]
    }
    
    response = requests.post(
        f'{api_url}/work-items/',
        headers=headers,
        json=payload,
        timeout=30
    )
    
    if response.status_code != 200:
        raise Exception(f'Failed to create Work Item: {response.text}')

Lambda Environment Variables:

WORKITEMS_API_URL=https://your-endpoint.sema4ai.work/tenants/xxx/api/v1/work-items
S4_API_KEY=<stored-in-secrets-manager>
AGENT_ID=document_processor
WEBHOOK_URL=https://your-app.com/webhooks/document-processed

S3 Event Configuration:

Go to your S3 bucket → Properties → Event notifications
Create notification for "All object create events"
Set destination to your Lambda function
Optionally filter by prefix/suffix (e.g., only .pdf files)

What the Worker Agent can do:

Extract data from invoices, receipts, contracts
Validate document structure and content
Compare against existing records
Route documents to appropriate systems
Generate summaries or reports

Trigger from Scheduled Tasks

Use case: Run recurring processes like daily reconciliations, weekly reports, monthly compliance checks, or batch processing. How it works:

A scheduler triggers at specified times
Scheduler creates Work Items for the scheduled task
Worker Agent processes according to schedule
Results are delivered via webhook or stored in your system

There are multiple ways to implement scheduled triggers, depending on your infrastructure. Here are the most common approaches:

AWS EventBridge (Lambda): Best for AWS-based, serverless architectures. A scheduled rule in EventBridge triggers a Lambda function that sends a request to the Work Items API. It's a "pay-per-execution" model with no infrastructure to manage.
Cron Job (Any Server): The most traditional and simple approach. A script is scheduled to run periodically using crontab on Linux/Mac or Task Scheduler on Windows. It works on any server and requires no specific cloud dependencies, but you must manage the server and handle failures manually.
Application-Based Scheduler: For applications that require more control or have their own internal scheduling needs. Libraries like node-cron (Node.js) or APScheduler (Python) are integrated directly into the application code to manage tasks. This gives you full control but requires the application to be continuously running.

Trigger from Web Application or ERPs

Use case: Create Work Items directly from your applications when users submit forms, place orders, or trigger workflows. This could be from your CRM, ERP, custom web app, or any system with API capabilities.

How it works:

User performs action in your app
Backend API handler creates Work Item
Worker Agent processes in background
Webhook notifies your app when complete

Setup requirements:

Backend server (Node.js, Python, etc.)
Sema4.ai API credentials
Webhook endpoint to receive completion notifications

Understanding Webhooks for Monitoring

Webhooks are HTTP callbacks that notify your application when Work Items reach specific statuses. They're essential for building responsive, event-driven systems.

Why use webhooks:

Real-time updates - Know immediately when processing completes
Error handling - React quickly to failures
Human-in-loop - Get notified when review is needed
Workflow continuation - Trigger next steps automatically
Monitoring - Track processing metrics

Webhook delivery:

Sent via HTTP POST to your specified URL
Includes Work Item ID, status, and payload
Retried up to 3 times with exponential backoff
Your endpoint should respond with 2xx status code

Next Steps

Now that you've learned how to create Work Items via the API, you can begin integrating Worker Agents into your business workflows. Explore advanced features such as file uploads, restarting or canceling Work Items, and manually setting Work Items to complete. For a comprehensive list of endpoints and capabilities, see the full API reference.

Publish and Deploy Worker Agents Reasoning Models