Anthropic · Claude Messages

# 安装依赖：python -m pip install requests
import base64
import json
import os
from pathlib import Path
from urllib.parse import quote
import requests

BASE_URL = "https://ai.kedang.net"
MODEL_ID = "MODEL_ID"  # 替换为账户可用、支持当前能力的模型


def request_json(path, body, extra_headers=None):
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    headers.update(extra_headers or {})
    response = requests.post(BASE_URL + path, headers=headers, json=body, timeout=120)
    response.raise_for_status()
    result = response.json()
    if result.get("error"):
        raise RuntimeError(result["error"])
    return result

def sse_lines(chunks):
    """增量解码 UTF-8，并跨网络分块正确识别 LF、CR 和 CRLF。"""
    import codecs
    decoder = codecs.getincrementaldecoder("utf-8")()
    line, skip_lf = [], False
    for chunk in chunks:
        for char in decoder.decode(chunk):
            if skip_lf:
                skip_lf = False
                if char == "\n":
                    continue
            if char in ("\r", "\n"):
                yield "".join(line)
                line = []
                skip_lf = char == "\r"
            else:
                line.append(char)
    decoder.decode(b"", final=True)  # 不接受 EOF 处截断的 UTF-8 字符。
    if line:
        yield "".join(line)


def iter_sse(lines):
    """只在空行处派发完整 SSE 帧。"""
    event, data_lines = "message", []
    first_line = True
    for line in lines:
        if first_line:
            line = line.removeprefix("\ufeff")
            first_line = False
        if line == "":
            if data_lines:
                yield event, "\n".join(data_lines)
            event, data_lines = "message", []
            continue
        if line.startswith(":"):
            continue  # 心跳或注释
        field, separator, value = line.partition(":")
        if separator and value.startswith(" "):
            value = value[1:]
        if field == "event":
            event = value or "message"
        elif field == "data":
            data_lines.append(value)
    # EOF 前没有空行的半帧不能当作完整事件；由下层检查终止事件。

def stream_text(events):
    completed = False
    for event, data in events:
        if data.strip() == "[DONE]":
            break
        item = json.loads(data)
        kind = item.get("type", event)
        if kind == "error" or item.get("error"):
            raise RuntimeError(item)
        if kind == "content_block_delta":
            delta = item.get("delta", {})
            if delta.get("type") == "text_delta":
                yield delta.get("text", "")
        elif kind == "message_delta":
            reason = item.get("delta", {}).get("stop_reason")
            if reason in ("max_tokens", "refusal", "model_context_window_exceeded"):
                raise RuntimeError("输出未完整完成：" + reason)
        elif kind == "message_stop":
            completed = True
            break
        # ping、思考块、签名增量不是面向用户的文本。
    if not completed:
        raise RuntimeError("连接在 message_stop 前结束。")

def main():
    path = "/v1/messages"
    body = {"model": MODEL_ID, "max_tokens": 1024, "messages": [{"role": "user", "content": "用三句话介绍杭州。"}], "stream": True}
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"],
               "Accept": "text/event-stream"}
    headers.update({"anthropic-version": "2023-06-01"})
    with requests.post(BASE_URL + path, headers=headers, json=body,
                       stream=True, timeout=(10, 120)) as response:
        response.raise_for_status()
        if "text/event-stream" not in response.headers.get("Content-Type", ""):
            raise RuntimeError("预期 SSE，请检查返回的 Content-Type 和错误信息。")
        events = iter_sse(sse_lines(response.iter_content(chunk_size=128)))
        for text in stream_text(events):
            print(text, end="", flush=True)
    print()

if __name__ == "__main__":
    main()

# 安装依赖：python -m pip install requests
import base64
import json
import os
from pathlib import Path
from urllib.parse import quote
import requests

BASE_URL = "https://ai.kedang.net"
MODEL_ID = "MODEL_ID"  # 替换为账户可用、支持当前能力的模型


def request_json(path, body, extra_headers=None):
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    headers.update(extra_headers or {})
    response = requests.post(BASE_URL + path, headers=headers, json=body, timeout=120)
    response.raise_for_status()
    result = response.json()
    if result.get("error"):
        raise RuntimeError(result["error"])
    return result

PARAMETERS = {
    "type": "object",
    "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
    "required": ["a", "b"],
    "additionalProperties": False,
}


def run_tool(name, args):
    # 工具名和参数都先校验；只执行应用主动注册的函数。
    if name != "add_numbers" or not isinstance(args, dict) or set(args) != {"a", "b"}:
        raise ValueError("未知工具或参数不匹配")
    if any(type(args[key]) not in (int, float) for key in ("a", "b")):
        raise ValueError("a 和 b 必须为数字")
    return {"sum": args["a"] + args["b"]}

def main():
    headers = {"anthropic-version": "2023-06-01"}
    tools = [{"name": "add_numbers", "description": "计算两个数字之和", "input_schema": PARAMETERS}]
    messages = [{"role": "user", "content": "请用工具计算 2 加 3。"}]
    first = request_json("/v1/messages", {"model": MODEL_ID, "max_tokens": 1024,
        "messages": messages, "tools": tools, "tool_choice": {"type": "auto"}}, headers)
    blocks = first["content"]
    calls = [block for block in blocks if block.get("type") == "tool_use"]
    if not calls:
        print(first)
        return
    messages.append({"role": "assistant", "content": blocks})  # 保留完整内容块与签名。
    results = []
    for call in calls:
        result = run_tool(call["name"], call["input"])
        results.append({"type": "tool_result", "tool_use_id": call["id"],
                        "content": json.dumps(result, ensure_ascii=False)})
    messages.append({"role": "user", "content": results})
    second = request_json("/v1/messages", {"model": MODEL_ID, "max_tokens": 1024,
        "messages": messages, "tools": tools, "tool_choice": {"type": "none"}}, headers)
    for block in second.get("content", []):
        if block.get("type") == "text":
            print(block["text"])

if __name__ == "__main__":
    main()

# 安装依赖：python -m pip install requests
import base64
import json
import os
from pathlib import Path
from urllib.parse import quote
import requests

BASE_URL = "https://ai.kedang.net"
MODEL_ID = "MODEL_ID"  # 替换为账户可用、支持当前能力的模型


def request_json(path, body, extra_headers=None):
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    headers.update(extra_headers or {})
    response = requests.post(BASE_URL + path, headers=headers, json=body, timeout=120)
    response.raise_for_status()
    result = response.json()
    if result.get("error"):
        raise RuntimeError(result["error"])
    return result

def main():
    image = base64.b64encode(Path("input.png").read_bytes()).decode("ascii")
    result = request_json("/v1/messages", {"model": MODEL_ID, "max_tokens": 1024,
        "messages": [{"role": "user", "content": [
            {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": image}},
            {"type": "text", "text": "请描述这张图片。"}
        ]}]}, {"anthropic-version": "2023-06-01"})
    for block in result.get("content", []):
        if block.get("type") == "text":
            print(block["text"])

if __name__ == "__main__":
    main()

# 安装依赖：python -m pip install requests
import base64
import json
import os
from pathlib import Path
from urllib.parse import quote
import requests

BASE_URL = "https://ai.kedang.net"
MODEL_ID = "MODEL_ID"  # 替换为账户可用、支持当前能力的模型


def request_json(path, body, extra_headers=None):
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    headers.update(extra_headers or {})
    response = requests.post(BASE_URL + path, headers=headers, json=body, timeout=120)
    response.raise_for_status()
    result = response.json()
    if result.get("error"):
        raise RuntimeError(result["error"])
    return result

SCHEMA = {
    "type": "object",
    "properties": {"city": {"type": "string"}, "days": {"type": "integer"}},
    "required": ["city", "days"],
    "additionalProperties": False,
}


def check_trip(value):
    # 业务侧仍校验结果，尤其要区分拒绝、截断和有效 JSON。
    if not isinstance(value, dict) or set(value) != {"city", "days"}:
        raise ValueError("返回对象不匹配 Schema")
    if not isinstance(value["city"], str) or type(value["days"]) is not int:
        raise ValueError("字段类型不匹配 Schema")
    return value

def main():
    # 以命名工具承接结构化对象；结果位于 tool_use.input，而非文本字符串。
    result = request_json("/v1/messages", {"model": MODEL_ID, "max_tokens": 1024,
        "messages": [{"role": "user", "content": "提取旅行信息：杭州，3 天。"}],
        "tools": [{"name": "save_trip", "description": "返回提取的旅行信息", "input_schema": SCHEMA}],
        "tool_choice": {"type": "tool", "name": "save_trip"}},
        {"anthropic-version": "2023-06-01"})
    if result.get("stop_reason") != "tool_use":
        raise RuntimeError("未得到完整工具输入")
    calls = [block for block in result["content"]
             if block.get("type") == "tool_use" and block.get("name") == "save_trip"]
    if len(calls) != 1:
        raise RuntimeError("预期一个 save_trip 结构化结果")
    print(check_trip(calls[0]["input"]))

if __name__ == "__main__":
    main()