Anthropic-compatible Messages

import anthropic
    import os

    client = anthropic.Anthropic(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # Replace {WorkspaceId} with your actual workspace ID. URLs vary by region.
        base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/apps/anthropic",
    )

    stream = client.messages.create(
        model="qwen3.8-max",
        max_tokens=1024,
        stream=True,
        messages=[
            {
                "role": "user",
                "content": "Give a brief introduction to artificial intelligence."
            }
        ],
        thinking={"type": "disabled"},
    )

    for chunk in stream:
        if chunk.type == "content_block_delta":
            if hasattr(chunk.delta, 'text'):
                print(chunk.delta.text, end="", flush=True)

import anthropic
    import os

    client = anthropic.Anthropic(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # Replace {WorkspaceId} with your actual workspace ID. URLs vary by region.
        base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/apps/anthropic",
    )

    stream = client.messages.create(
        model="qwen3.8-max",
        max_tokens=2048,
        stream=True,
        thinking={
            "type": "enabled",
            "budget_tokens": 1024
        },
        messages=[
            {
                "role": "user",
                "content": "Analyze the future prospects of quantum computing."
            }
        ]
    )

    for chunk in stream:
        if chunk.type == "content_block_delta":
            if hasattr(chunk.delta, 'thinking'):
                print(chunk.delta.thinking, end="", flush=True)
            elif hasattr(chunk.delta, 'text'):
                print(chunk.delta.text, end="", flush=True)

import anthropic
    import os

    client = anthropic.Anthropic(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # Replace {WorkspaceId} with your actual workspace ID. URLs vary by region.
        base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/apps/anthropic",
    )

    stream = client.messages.create(
        model="qwen3.8-max",
        max_tokens=1024,
        stream=True,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "url",
                            "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20250414/mqqmiy/animal_01.jpg",
                        },
                    },
                    {
                        "type": "text",
                        "text": "Describe the content of this image."
                    },
                ],
            }
        ],
        thinking={"type": "disabled"},
    )

    for chunk in stream:
        if chunk.type == "content_block_delta":
            if hasattr(chunk.delta, 'text'):
                print(chunk.delta.text, end="", flush=True)

import anthropic
    import os

    client = anthropic.Anthropic(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # Replace {WorkspaceId} with your actual workspace ID. URLs vary by region.
        base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/apps/anthropic",
    )

    stream = client.messages.create(
        model="qwen3.8-max",
        max_tokens=1024,
        stream=True,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "video",
                        "source": {
                            "type": "url",
                            "url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20251208/zpupby/3e81ef38-98f0-4d55-bbb6-259334ca18d0.mp4",
                        },
                    },
                    {
                        "type": "text",
                        "text": "Describe the content of this video."
                    },
                ],
            }
        ],
        thinking={"type": "disabled"},
    )

    for chunk in stream:
        if chunk.type == "content_block_delta":
            if hasattr(chunk.delta, 'text'):
                print(chunk.delta.text, end="", flush=True)

import anthropic
    import os

    client = anthropic.Anthropic(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # Replace {WorkspaceId} with your actual workspace ID. URLs vary by region.
        base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/apps/anthropic",
    )

    tools = [
        {
            "name": "get_weather",
            "description": "Get weather information for a specified city",
            "input_schema": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name"
                    }
                },
                "required": ["city"]
            }
        }
    ]

    message = client.messages.create(
        model="qwen3.8-max",
        max_tokens=1024,
        tools=tools,
        messages=[
            {
                "role": "user",
                "content": "What's the weather like in Hangzhou today?"
            }
        ]
    )

    print(message.content)

import anthropic
    import os

    client = anthropic.Anthropic(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # Replace {WorkspaceId} with your actual workspace ID. URLs vary by region.
        base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/apps/anthropic",
    )

    # Simulate code repository content. Must reach minimum cacheable length (1024 tokens)
    long_text_content = "<Your Code Here>" * 400

    def get_completion(user_input):
        response = client.messages.create(
            # Choose a model that supports prompt caching
            model="qwen3.8-max",
            max_tokens=1024,
            system=[
                {
                    "type": "text",
                    "text": long_text_content,
                    # Add cache_control on a text block to mark a cache breakpoint. Can also be placed on content blocks in the messages array
                    "cache_control": {"type": "ephemeral"},
                }
            ],
            messages=[
                {"role": "user", "content": user_input},
            ],
        )
        return response

    # First request: Create cache
    first = get_completion("What does this code do?")
    print(f"Cache creation tokens: {first.usage.cache_creation_input_tokens}")
    print(f"Cache read tokens: {first.usage.cache_read_input_tokens}")
    print("=" * 20)
    # Second request: Same long content, different question -> Cache hit
    second = get_completion("How can this code be optimized?")
    print(f"Cache creation tokens: {second.usage.cache_creation_input_tokens}")
    print(f"Cache read tokens: {second.usage.cache_read_input_tokens}")

import anthropic
    import os

    client = anthropic.Anthropic(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        # Replace {WorkspaceId} with your actual workspace ID. URLs vary by region.
        base_url="https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/apps/anthropic",
    )

    message = client.messages.create(
        model="deepseek-v4-pro",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "Extract key info from this email: John Smith (john@example.com) is interested in the Enterprise plan and wants to schedule a demo for next Tuesday at 2pm."
            }
        ],
        extra_body={
            "output_config": {
                "format": {
                    "type": "json_schema",
                    "schema": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "email": {"type": "string"},
                            "plan_interest": {"type": "string"},
                            "demo_requested": {"type": "boolean"}
                        },
                        "required": ["name", "email", "plan_interest", "demo_requested"],
                        "additionalProperties": False
                    }
                }
            }
        },
    )

    # deepseek-v4-pro returns a thinking block; find the text content block
    text_block = next(block for block in message.content if block.type == "text")
    print(text_block.text)