{
  "stream": true,
  "content": "请用一句话告诉我iDA是什么产品",
  "metadata": { "agentId": 17368, "enableKnowledgeQA": true, "enableRunningStepOutput": true }
}

data: {"task":{"kind":"task","id":"8d06d22ced4d4b449cc414c9283cde56","contextId":"e3dda1cf238147b094fb7aa9bfea0fee","status":{"state":"submitted","timestamp":"2026-05-21T14:07:33.304538"},"metadata":{...}}}

data: {"statusUpdate":{"kind":"status-update","taskId":"8d06d22ced4d4b449cc414c9283cde56","status":{"state":"working","timestamp":"2026-05-21T14:07:34.500000"},"final":false,"metadata":{...}}}

data: {"artifactUpdate":{"kind":"artifact-update","taskId":"8d06d22ced4d4b449cc414c9283cde56","artifact":{"artifactId":"...","name":"思考推理","parts":[{"kind":"text","text":"iDA（intelligent Data Analysis）是一款智能数据分析产品，通过集成AI技术帮助用户更高效地进行数据探索、分析和可视化。"}],"metadata":{"type":"deep_research_running_step","stepType":"executor_think"}},"lastChunk":true}}

data: {"statusUpdate":{"kind":"status-update","taskId":"8d06d22ced4d4b449cc414c9283cde56","status":{"state":"completed","timestamp":"2026-05-21T14:09:58.160050"},"final":true,"metadata":{...}}}

data: [DONE]

第二轮请求时，把上一轮 taskDetail 返回的 task.contextId（同时也可一并带上 metadata.sessionId）回传：

{
  "stream": true,
  "content": "上一轮的结果再帮我换成英文输出",
  "contextId": "e3dda1cf238147b094fb7aa9bfea0fee",
  "metadata": { "agentId": 17368, "sessionId": 5033, "enableKnowledgeQA": true, "enableRunningStepOutput": true }
}

import json, time, requests

DOMAIN = "https://ida.bytedance.net"
CLIENT_ID = "<your_client_id>"
CLIENT_SECRET = "<your_client_secret>"
PROXY_USER = "admin"
AGENT_ID = 17368  # CN 通用智能体


def fetch_jwt() -> str:
    r = requests.post(
        f"{DOMAIN}/bi/aeolus/api/v3/openapi/jwtToken",
        json={"metadata": {"clientId": CLIENT_ID, "clientSecret": CLIENT_SECRET, "proxyUser": PROXY_USER, "expire": 259200}},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()["data"]["jwtToken"]


def headers(jwt: str) -> dict:
    return {
        "Authorization": f"Bearer {jwt}",
        "X-DataAgent-User": PROXY_USER,
        "Content-Type": "application/json",
        "X-request-id": "demo-uuid",
    }


def ask_stream(question: str) -> str:
    jwt = fetch_jwt()
    body = {"stream": True, "content": question,
            "metadata": {"agentId": AGENT_ID, "enableKnowledgeQA": True, "enableRunningStepOutput": True}}
    final_answer = ""
    with requests.post(
        f"{DOMAIN}/dataAgent/llm/openApi/v2/deepresearch/executeTask",
        headers={**headers(jwt), "Accept": "text/event-stream"},
        json=body, stream=True, timeout=600,
    ) as resp:
        resp.raise_for_status()
        for raw in resp.iter_lines(decode_unicode=True):
            if not raw or not raw.startswith("data:"):
                continue
            payload = raw[len("data:"):].strip()
            if payload == "[DONE]":
                break
            evt = json.loads(payload)
            au = evt.get("artifactUpdate")
            if au and au.get("lastChunk"):
                art = au.get("artifact") or {}
                if (art.get("metadata") or {}).get("stepType") == "executor_think":
                    final_answer = "\n".join(p.get("text", "") for p in art.get("parts", []) if p.get("kind") == "text")
    return final_answer


def ask_blocking(question: str, poll_interval: int = 3, timeout: int = 600) -> dict:
    jwt = fetch_jwt()
    body = {"stream": False, "content": question,
            "metadata": {"agentId": AGENT_ID, "enableKnowledgeQA": True, "enableRunningStepOutput": False}}
    r = requests.post(f"{DOMAIN}/dataAgent/llm/openApi/v2/deepresearch/executeTask",
                      headers=headers(jwt), json=body, timeout=120)
    r.raise_for_status()
    task_id = r.json()["data"]["taskId"]

    deadline = time.time() + timeout
    while time.time() < deadline:
        s = requests.get(f"{DOMAIN}/dataAgent/llm/openApi/v2/deepresearch/taskStatus",
                         headers=headers(jwt), params={"id": task_id}, timeout=30).json()
        state = (s["data"]["status"] or {}).get("state")
        if state in ("completed", "failed"):
            break
        time.sleep(poll_interval)

    detail = requests.get(f"{DOMAIN}/dataAgent/llm/openApi/v2/deepresearch/taskDetail",
                          headers=headers(jwt), params={"id": task_id}, timeout=30).json()
    artifacts = detail["data"]["task"].get("artifacts") or []
    final_answer = ""
    for art in artifacts:
        if (art.get("metadata") or {}).get("stepType") == "executor_think":
            final_answer = "\n".join(p.get("text", "") for p in art.get("parts", []) if p.get("kind") == "text")
    return {"task_id": task_id, "answer": final_answer, "detail": detail}


if __name__ == "__main__":
    print(ask_stream("请用一句话告诉我iphone是什么产品"))