How to integrate any Dify app — chatbot, workflow, or agent — into external applications

Why the Dify API matters

Dify's visual editor is great for building, but the real value is embedding those apps into your own products. Dify exposes every app type — chatbot, workflow, agent — as a REST API. You build in Dify, call it from anywhere.

This guide covers authentication, the four API types, streaming vs blocking responses, conversation persistence, error handling, and rate limiting.

Authentication

Every API call requires a Bearer token. Each Dify app has its own API key — find it under your app's API Access settings.

# All requests use this header
Authorization: Bearer YOUR_APP_API_KEY
 
API keys are per-app, not per-account. If you have five Dify apps, you have five different API keys. Do not confuse them.

The four API types

App type Endpoint prefix Use case
Chatbot / Chatflow /v1/chat-messages Conversational back-and-forth
Completion (legacy) /v1/completion-messages Single-turn text generation
Workflow /v1/workflows/run Run a Workflow app
Agent /v1/chat-messages Same endpoint as Chatbot

Chat API: blocking vs streaming

The most common call. Use response_mode: blocking for simple integrations and response_mode: streaming for real-time UIs.

Blocking call

import requests
 
API_KEY = 'app-xxxxxxxxxxxx'
BASE_URL = 'https://api.dify.ai/v1'
 
def chat(message: str, conversation_id: str = '') -> dict:
    resp = requests.post(
        f'{BASE_URL}/chat-messages',
        headers={'Authorization': f'Bearer {API_KEY}'},
        json={
            'inputs': {},
            'query': message,
            'response_mode': 'blocking',
            'conversation_id': conversation_id,
            'user': 'user-123',
        }
    )
    resp.raise_for_status()
    data = resp.json()
    return {
        'answer': data['answer'],
        'conversation_id': data['conversation_id'],
    }
 
# First turn
result = chat('What is the refund policy?')
print(result['answer'])
 
# Continue the conversation
result2 = chat('What about international orders?', result['conversation_id'])

Streaming call

import requests
import json
 
def chat_stream(message: str, conversation_id: str = ''):
    """Yields text chunks as they arrive."""
    resp = requests.post(
        f'{BASE_URL}/chat-messages',
        headers={'Authorization': f'Bearer {API_KEY}'},
        json={
            'inputs': {},
            'query': message,
            'response_mode': 'streaming',
            'conversation_id': conversation_id,
            'user': 'user-123',
        },
        stream=True
    )
    resp.raise_for_status()
    for line in resp.iter_lines():
        if line and line.startswith(b'data: '):
            payload = json.loads(line[6:])
            event = payload.get('event')
            if event == 'message':
                yield payload.get('answer', '')
            elif event == 'message_end':
                return
 
for chunk in chat_stream('Explain quantum entanglement simply.'):
    print(chunk, end='', flush=True)

Workflow API

Running a Workflow app is slightly different — you pass named inputs instead of a query string.

def run_workflow(inputs: dict) -> dict:
    resp = requests.post(
        f'{BASE_URL}/workflows/run',
        headers={'Authorization': f'Bearer {API_KEY}'},
        json={
            'inputs': inputs,       # matches your Workflow's input variables
            'response_mode': 'blocking',
            'user': 'user-123',
        }
    )
    resp.raise_for_status()
    return resp.json()['data']['outputs']
 
result = run_workflow({'document_url': 'https://example.com/contract.pdf'})
print(result['summary'])

Conversation persistence

For chatbot/agent apps, Dify automatically maintains conversation history server-side. Your only job is to pass the conversation_id back on every subsequent turn.

class DifyChat:
    def __init__(self, user_id: str):
        self.user_id = user_id
        self.conversation_id = ''
 
    def send(self, message: str) -> str:
        result = chat(message, self.conversation_id)
        self.conversation_id = result['conversation_id']  # persist
        return result['answer']
 
session = DifyChat('user-456')
print(session.send('Hi, I need help with my order'))
print(session.send('Order number is 12345'))  # context preserved

Error handling

HTTP status Meaning What to do
400 Invalid request body Check your inputs dict matches the app's variables
401 Bad API key Re-check the key — it is per-app, not per-account
404 App not found Confirm the app is published, not just saved as draft
429 Rate limited Back off with exponential retry; consider caching responses
500 Dify server error Retry once; if persistent, check Dify status page 
import time
 
def chat_with_retry(message: str, conversation_id: str = '', max_retries: int = 3) -> dict:
    for attempt in range(max_retries):
        try:
            return chat(message, conversation_id)
        except requests.HTTPError as e:
            if e.response.status_code == 429:
                wait = 2 ** attempt
                print(f'Rate limited, waiting {wait}s...')
                time.sleep(wait)
            elif e.response.status_code >= 500:
                time.sleep(1)
            else:
                raise  # 400/401/404 are not retryable
    raise RuntimeError('Max retries exceeded')

Rate limits and multi-tenant patterns

Dify Cloud enforces rate limits per API key. For multi-tenant apps where each customer should be isolated, pass a unique user parameter per request — Dify uses this for conversation isolation and the analytics dashboard.

The user field in the request body is your tenant/user identifier. It does not affect the LLM call but is used by Dify's analytics and helps you audit usage per customer in the Logs tab.

Self-hosted rate limiting

On self-hosted Dify you can configure Nginx rate limiting upstream. The Dify app itself does not enforce rate limits on self-hosted deployments by default.

Passing file inputs

Some Dify apps accept file uploads. You can pass files as base64 or upload them first with the file upload endpoint.

# Upload a file first
def upload_file(path: str) -> str:
    with open(path, 'rb') as f:
        resp = requests.post(
            f'{BASE_URL}/files/upload',
            headers={'Authorization': f'Bearer {API_KEY}'},
            files={'file': f},
            data={'user': 'user-123'}
        )
    resp.raise_for_status()
    return resp.json()['id']
 
# Then reference the file ID in your chat or workflow call
file_id = upload_file('contract.pdf')
chat_with_file = {
    'inputs': {},
    'query': 'Summarise this contract',
    'files': [{'type': 'document', 'transfer_method': 'local_file', 'upload_file_id': file_id}],
    'response_mode': 'blocking',
    'user': 'user-123',
}

Quick checklist

  • API key is per-app — check you are using the right one
  • Use response_mode: streaming for any user-facing UI
  • Store and pass conversation_id on every turn for chat apps
  • The user field is your tenant identifier — make it unique per customer
  • Publish your app (not just save) before the API will respond
  • Handle 429 with exponential backoff