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)