Python 调用 Coze/Dify：认证、输入契约和流式事件

python -m pip install "httpx>=0.27,<1"
export DIFY_BASE_URL="https://api.dify.ai/v1"
export DIFY_API_KEY="替换为已发布应用的密钥"

import json
import os
from collections.abc import Iterator

import httpx


def iter_sse_data(lines: Iterator[str]) -> Iterator[dict]:
    data_lines: list[str] = []
    for line in lines:
        if line == "":
            if data_lines:
                yield json.loads("\n".join(data_lines))
                data_lines.clear()
            continue
        if line.startswith("data:"):
            data_lines.append(line[5:].lstrip())
    if data_lines:
        yield json.loads("\n".join(data_lines))


def run_dify_workflow(inputs: dict, user_id: str) -> Iterator[dict]:
    base_url = os.environ["DIFY_BASE_URL"].rstrip("/")
    token = os.environ["DIFY_API_KEY"]
    payload = {"inputs": inputs, "response_mode": "streaming", "user": user_id}

    timeout = httpx.Timeout(connect=5, read=120, write=10, pool=5)
    with httpx.Client(timeout=timeout) as client:
        with client.stream(
            "POST",
            f"{base_url}/workflows/run",
            headers={"Authorization": f"Bearer {token}"},
            json=payload,
        ) as response:
            response.raise_for_status()
            for event in iter_sse_data(response.iter_lines()):
                event_type = event.get("event", "unknown")
                yield {"type": event_type, "data": event}
                if event_type in {"workflow_finished", "error"}:
                    return


for item in run_dify_workflow(
    {"device_id": "AX-3", "status": "offline", "temperature": 31},
    user_id="user-17",
):
    print(json.dumps(item, ensure_ascii=False))

{"type": "started|progress|completed|failed", "run_id": "...", "payload": {}}