### Initialize CozeLoop Client with PAT Authentication

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Create an explicit Client instance using Personal Access Token (PAT) authentication. Workspace ID and API Token can be read from environment variables if not provided. This example also shows how to start, set input/output, and finish a trace span, and finally close the client.

```python
import cozeloop

# PAT authentication — reads workspace ID and token from env vars if not provided
client = cozeloop.new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",   # or set COZELOOP_WORKSPACE_ID
    api_token="pat_xxxxxxxxxxxxxxxx",       # or set COZELOOP_API_TOKEN

    # JWT OAuth authentication (alternative to api_token)
    # jwt_oauth_client_id="your_client_id",
    # jwt_oauth_private_key="-----BEGIN RSA PRIVATE KEY-----\n...",
    # jwt_oauth_public_key_id="your_public_key_id",

    timeout=3,                          # HTTP read timeout in seconds (default: 3)
    upload_timeout=30,                  # File/large-payload upload timeout (default: 30)
    ultra_large_report=False,           # Upload >1MB inputs/outputs as files (default: False)
    prompt_cache_max_count=100,         # Max prompts cached locally (default: 100)
    prompt_cache_refresh_interval=600,  # Cache TTL in seconds (default: 60)
    prompt_trace=False,                 # Emit spans for get_prompt/prompt_format (default: False)
)

span = client.start_span("root", "custom")
span.set_input("hello")
span.set_output("world")
span.finish()

client.close()  # Always close before program exit to flush queued spans
```

--------------------------------

### Install Project Dependencies with Poetry

Source: https://github.com/coze-dev/cozeloop-python/blob/main/CONTRIBUTING.md

Use this command to install all necessary project dependencies. Ensure Poetry is installed and configured correctly.

```bash
poetry install
```

--------------------------------

### Use Module-Level Default Client with Environment Variables

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Utilize the module-level API for zero-configuration usage by setting environment variables for workspace ID and API token. The SDK automatically creates and manages a default client. This example demonstrates starting a span, setting input/output, finishing the span, and explicitly closing the client (though `atexit` can handle this automatically).

```python
import os
import cozeloop

os.environ["COZELOOP_WORKSPACE_ID"] = "7421xxxxxxxxxxxxxxxx"
os.environ["COZELOOP_API_TOKEN"]     = "pat_xxxxxxxxxxxxxxxx"

# No explicit client needed — module-level functions use the auto-created default client
span = cozeloop.start_span("my_span", "custom")
span.set_input({"query": "What is AI?"})
span.set_output({"answer": "AI stands for Artificial Intelligence."})
span.finish()

cozeloop.close()  # Explicit close; alternatively atexit handles it automatically
```

--------------------------------

### Create a Root Trace Span

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Start a new trace span using the module-level `cozeloop.start_span()` function. This span is automatically linked as a child of the most recent span in the current thread's context and begins timing immediately. This example also shows setting user ID and thread ID for the span.

```python
import cozeloop
from cozeloop.spec.tracespec import ModelInput, ModelMessage, ModelCallOption, CALL_OPTIONS

# Root span
root = cozeloop.start_span("pipeline", "custom")
root.set_user_id("user_001")
root.set_thread_id("thread_abc")

```

--------------------------------

### Install CozeLoop Python SDK

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Install the CozeLoop Python SDK using pip. This command installs the latest stable version.

```bash
pip install CozeLoop
```

--------------------------------

### Get and Format Prompt with CozeLoop SDK

Source: https://github.com/coze-dev/cozeloop-python/blob/main/README.md

Retrieve a prompt using its key and then format it with provided variables. This is useful for dynamic prompt generation.

```python
def main():
    prompt = cozeloop.get_prompt(prompt_key="your_prompt_key")
    messages = cozeloop.prompt_format(
        prompt,
        {"var1": "your content"},
    )
```

--------------------------------

### Report Trace with CozeLoop SDK

Source: https://github.com/coze-dev/cozeloop-python/blob/main/README.md

Use this snippet to start a trace span, set input and output, and finish the span. Ensure cozeloop.close() is called to finalize the trace.

```python
def main():
    span = cozeloop.start_span("root", "custom")

    span.set_input("Hello") 
    span.set_output("World") 
	
    span.finish()
	
    cozeloop.close()
```

--------------------------------

### start_span(name, span_type, ...) - Create a new trace span

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Creates a Span and automatically links it as a child of the most recent span in the current thread's context. The span starts timing immediately. Available at both module level and via an explicit client instance.

```APIDOC
## start_span(name, span_type, ...)

### Description
Creates a `Span` and automatically links it as a child of the most recent span in the current thread's context. The span starts timing immediately. On the module level, `cozeloop.start_span()`; on an explicit client, `client.start_span()`.

### Method
`cozeloop.start_span(name: str, span_type: str, ...)` or `client.start_span(name: str, span_type: str, ...)`

### Parameters
- **name** (str) - The name of the span.
- **span_type** (str) - The type of the span (e.g., "custom", "llm", "tool").
- **kwargs** - Additional keyword arguments can be passed to set span attributes.

### Usage
Spans can be created using the module-level function or an explicit client instance.

### Request Example
```python
import cozeloop
from cozeloop.spec.tracespec import ModelInput, ModelMessage, ModelCallOption, CALL_OPTIONS

# Using module-level function
root = cozeloop.start_span("pipeline", "custom")
root.set_user_id("user_001")
root.set_thread_id("thread_abc")

# Using explicit client instance
# client = cozeloop.new_client(...)
# root = client.start_span("pipeline", "custom")
# root.set_user_id("user_001")
# root.set_thread_id("thread_abc")
```

### Response
- **Span Object** - A `Span` object that can be used to record events, inputs, outputs, and finish the trace.
```

--------------------------------

### Execute Prompts with CozeLoop SDK (Sync & Async)

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Demonstrates synchronous, asynchronous non-streaming, and asynchronous streaming execution of prompts using the CozeLoop client. Ensure the client is properly initialized before use.

```python
stream_reader = client.execute_prompt(
    prompt_key="ptaas_demo",
    variable_vals={"topic": "AI", "user_request": "What is GPT?"},
    stream=True,
)
for chunk in stream_reader:
    if chunk.message and chunk.message.content:
        print(chunk.message.content, end="", flush=True)
print()
```

```python
async def run_async():
    result = await client.aexecute_prompt(
        prompt_key="ptaas_demo",
        version="0.0.1",
        variable_vals={"topic": "AI", "user_request": "What is a neural network?"},
        stream=False,
    )
    print(result.message)

asyncio.run(run_async())
```

```python
async def run_async_stream():
    stream_reader = await client.aexecute_prompt(
        prompt_key="ptaas_demo",
        variable_vals={"topic": "AI", "user_request": "Describe deep learning."},
        stream=True,
    )
    async for chunk in stream_reader:
        if chunk.message and chunk.message.content:
            print(chunk.message.content, end="", flush=True)
    print()

asyncio.run(run_async_stream())
```

--------------------------------

### Execute Prompt (Sync and Async)

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Demonstrates how to execute prompts using the CozeLoop SDK, both synchronously and asynchronously, with support for streaming responses.

```APIDOC
## Execute Prompt

### Description
Execute a prompt with the CozeLoop client. Supports both synchronous and asynchronous execution, with an option for streaming responses.

### Methods
- `client.execute_prompt(...)` (sync)
- `client.aexecute_prompt(...)` (async)

### Parameters
- `prompt_key` (string): The key of the prompt to execute.
- `variable_vals` (dict): A dictionary of variable values for the prompt.
- `stream` (bool): Whether to stream the response. Defaults to `False`.
- `version` (string, optional): The version of the prompt to use.

### Request Example (Sync Streaming)
```python
stream_reader = client.execute_prompt(
    prompt_key="ptaas_demo",
    variable_vals={"topic": "AI", "user_request": "What is GPT?"},
    stream=True,
)
for chunk in stream_reader:
    if chunk.message and chunk.message.content:
        print(chunk.message.content, end="", flush=True)
print()
```

### Request Example (Async Non-Streaming)
```python
async def run_async():
    result = await client.aexecute_prompt(
        prompt_key="ptaas_demo",
        version="0.0.1",
        variable_vals={"topic": "AI", "user_request": "What is a neural network?"},
        stream=False,
    )
    print(result.message)

asyncio.run(run_async())
```

### Request Example (Async Streaming)
```python
async def run_async_stream():
    stream_reader = await client.aexecute_prompt(
        prompt_key="ptaas_demo",
        variable_vals={"topic": "AI", "user_request": "Describe deep learning."},
        stream=True,
    )
    async for chunk in stream_reader:
        if chunk.message and chunk.message.content:
            print(chunk.message.content, end="", flush=True)
    print()

asyncio.run(run_async_stream())
```

### Cleanup
```python
client.close()
```
```

--------------------------------

### Pipeline Execution with Environment Variables

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Demonstrates a basic pipeline execution, including setting necessary environment variables for workspace and API token. It also shows how to handle asynchronous generators and flush the cozeloop client.

```python
import asyncio
import cozeloop
from cozeloop.decorators import observe

@observe()
def pipeline(user_input: str) -> str:
    cleaned   = preprocess(user_input)
    response  = call_llm(cleaned)
    return response

if __name__ == "__main__":
    import os
    os.environ["COZELOOP_WORKSPACE_ID"] = "7421xxxxxxxxxxxxxxxx"
    os.environ["COZELOOP_API_TOKEN"]     = "pat_xxxxxxxxxxxxxxxx"

    result = pipeline("  Hello World  ")
    print(result)

    # For async generator
    async def run():
        async for item in (await async_fetch("https://example.com"),):
            pass
        tokens = list(stream_tokens("hello world foo"))
        print(tokens)
    asyncio.run(run())

    cozeloop.flush()
```

--------------------------------

### cozeloop.new_client() - Create an explicit client instance

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Creates and returns a Client instance with full configuration control. Calling new_client() with identical arguments returns the same cached instance. The first client created is automatically set as the default client for module-level functions.

```APIDOC
## cozeloop.new_client()

### Description
Creates and returns a `Client` instance with full configuration control. Calling `new_client()` with identical arguments returns the same cached instance. The first client created is automatically set as the default client for module-level functions.

### Method
`cozeloop.new_client()`

### Parameters
- **workspace_id** (str) - Optional - Your CozeLoop workspace ID. Reads from `COZELOOP_WORKSPACE_ID` environment variable if not provided.
- **api_token** (str) - Optional - Personal Access Token (PAT) for authentication. Reads from `COZELOOP_API_TOKEN` environment variable if not provided. Alternatively, JWT OAuth credentials can be used.
- **jwt_oauth_client_id** (str) - Optional - JWT OAuth app client ID.
- **jwt_oauth_private_key** (str) - Optional - JWT OAuth private key.
- **jwt_oauth_public_key_id** (str) - Optional - JWT OAuth public key ID.
- **timeout** (int) - Optional - HTTP read timeout in seconds (default: 3).
- **upload_timeout** (int) - Optional - File/large-payload upload timeout (default: 30).
- **ultra_large_report** (bool) - Optional - Upload >1MB inputs/outputs as files (default: False).
- **prompt_cache_max_count** (int) - Optional - Max prompts cached locally (default: 100).
- **prompt_cache_refresh_interval** (int) - Optional - Cache TTL in seconds (default: 60).
- **prompt_trace** (bool) - Optional - Emit spans for get_prompt/prompt_format (default: False).

### Request Example
```python
import cozeloop

client = cozeloop.new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
    timeout=3,
    prompt_cache_max_count=100,
    prompt_cache_refresh_interval=600,
)

span = client.start_span("root", "custom")
span.set_input("hello")
span.set_output("world")
span.finish()

client.close()
```

### Response
- **Client Instance** - A configured CozeLoop client instance.
```

--------------------------------

### CozeLoop Data Models for Tracing

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Imports and demonstrates the usage of core data models for structured input and output in model spans, including multimodal message support.

```python
from cozeloop.entities.prompt import (
    Prompt,
    Message,
    Role,
    ExecuteResult,
    TokenUsage,
    LLMConfig,
    PromptTemplate,
    ContentPart,
)

from cozeloop.spec.tracespec import (
    ModelInput,
    ModelOutput,
    ModelMessage,
    ModelCallOption,
    ModelImageURL,
)
```

```python
from cozeloop.spec.tracespec import ModelMessagePart, ModelMessagePartType

multimodal_input = ModelInput(messages=[
    ModelMessage(
        role="user",
        parts=[
            ModelMessagePart(type=ModelMessagePartType.TEXT, text="Describe this image:"),
            ModelMessagePart(
                type=ModelMessagePartType.IMAGE,
                image_url=ModelImageURL(url="https://example.com/photo.jpg")
            ),
        ]
    )
])

span = cozeloop.start_span("vision_call", "model")
span.set_input(multimodal_input)
span.set_output("The image shows a mountain landscape at sunset.")
span.finish()
```

--------------------------------

### Span Interface - Business Tags and Deployment Environment

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Add arbitrary business tags and set the deployment environment for a span.

```python
# Arbitrary business tags
span.set_tags({"experiment_id": "exp_v2", "region": "us-east-1"})

# Deployment environment
span.set_deployment_env("production")
```

--------------------------------

### Execute a prompt against an LLM (synchronous)

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Executes a managed prompt on the CozeLoop backend synchronously, filling variables and calling the configured LLM. Returns ExecuteResult.

```APIDOC
## execute_prompt(prompt_key, ...)

### Description
Execute a managed prompt synchronously against an LLM.

### Method Signature
`execute_prompt(prompt_key: str, ..., stream: Literal[False] = False, timeout: Optional[int] = None) -> ExecuteResult`

### Parameters
- **prompt_key** (str) - Required - The key of the prompt to execute.
- **version** (str, optional) - The version of the prompt.
- **label** (str, optional) - The label of the prompt.
- **variable_vals** (Dict[str, Any], optional) - Values for prompt variables.
- **messages** (List[Message], optional) - Additional messages to include in the prompt.
- **stream** (Literal[False]) - Must be False for synchronous, non-streaming execution.
- **timeout** (int, optional) - Execution timeout in seconds.

### Returns
- `ExecuteResult` - An object containing the LLM's response message, finish reason, and token usage.

### Example
```python
import cozeloop
from cozeloop.entities.prompt import Message, Role, ExecuteResult

client = cozeloop.new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
)

result: ExecuteResult = client.execute_prompt(
    prompt_key="ptaas_demo",
    version="0.0.1",
    variable_vals={
        "topic":        "artificial intelligence",
        "user_request": "explain what is machine learning",
    },
    messages=[Message(role=Role.USER, content="Keep the answer brief.")],
    stream=False,
    timeout=60,
)
print(result.message)
print(result.finish_reason)
print(result.usage)

client.close()
```
```

--------------------------------

### Span Interface - Core Operations

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Demonstrates the core methods for creating and manipulating spans, including setting input, output, and errors.

```APIDOC
## Span Interface — Full list of setters

A `Span` object exposes the following methods after being returned by `start_span()`.

```python
span = cozeloop.start_span("my_op", "custom")

# Core input / output / error
span.set_input({"messages": [{"role": "user", "content": "Hello"}]})
span.set_output("Hi there!")
span.set_error(ValueError("something went wrong"))
span.set_status_code(-1)          # Non-zero = error

# Correlation identifiers
span.set_user_id("user_42")
span.set_message_id("msg_99")
span.set_thread_id("thread_7")
span.set_log_id("log_xyz")

# Promote identifiers to baggage (propagated to all child spans automatically)
span.set_user_id_baggage("user_42")
span.set_thread_id_baggage("thread_7")

# Model-specific fields
span.set_model_provider("anthropic")
span.set_model_name("claude-3-5-sonnet")
span.set_input_tokens(800)
span.set_output_tokens(400)
span.set_start_time_first_resp(1700000000000000)  # first-token latency anchor (μs)

# Arbitrary business tags
span.set_tags({"experiment_id": "exp_v2", "region": "us-east-1"})

# Baggage — inherited by downstream child spans
span.set_baggage({"tenant": "acme-corp"})

# Deployment environment
span.set_deployment_env("production")

# Context manager usage — finish() called automatically on __exit__
with cozeloop.start_span("guarded_op", "custom") as s:
    s.set_input("safe input")
    # span.finish() is called even if an exception is raised

# Cross-process / cross-service context propagation
headers = span.to_header()
# {"X-Cozeloop-Traceparent": "...", "X-Cozeloop-Tracestate": "..."}

# On the receiving service, reconstruct parent context from incoming headers
parent_ctx = cozeloop.get_span_from_header(headers)
child = cozeloop.start_span("downstream_op", "custom", child_of=parent_ctx)
child.finish()

span.finish()
```
```

--------------------------------

### Manually Create and Configure a Model Span

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Use this to manually create a span for a model call, setting provider, name, input, output tokens, and tags. Ensure `cozeloop.flush()` is called to report spans.

```python
model_span = cozeloop.start_span("llm_call", "model")
model_span.set_model_provider("openai")
model_span.set_model_name("gpt-4o")
model_span.set_input(ModelInput(
    messages=[
        ModelMessage(role="system", content="You are a helpful assistant."),
        ModelMessage(role="user",   content="Explain transformers."),
    ]
))
model_span.set_input_tokens(512)
model_span.set_output_tokens(256)
model_span.set_start_time_first_resp(1700000000123456)  # microseconds
model_span.set_tags({
    CALL_OPTIONS: ModelCallOption(temperature=0.7, max_tokens=1024, top_p=0.95)
})
model_span.set_output("Transformers are a neural network architecture...")
model_span.finish()

root.finish()
cozeloop.flush()  # Block until all queued spans are reported
```

--------------------------------

### Module-level default client - Zero-config usage

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

When any module-level function is called without a prior new_client(), a default client is auto-created from environment variables and registered with atexit for graceful shutdown.

```APIDOC
## Module-level default client

### Description
When any module-level function is called without a prior `new_client()`, a default client is auto-created from environment variables and registered with `atexit` for graceful shutdown.

### Method
Module-level functions like `cozeloop.start_span()` and `cozeloop.close()`.

### Usage
Ensure environment variables `COZELOOP_WORKSPACE_ID` and `COZELOOP_API_TOKEN` are set.

### Request Example
```python
import os
import cozeloop

os.environ["COZELOOP_WORKSPACE_ID"] = "7421xxxxxxxxxxxxxxxx"
os.environ["COZELOOP_API_TOKEN"]     = "pat_xxxxxxxxxxxxxxxx"

span = cozeloop.start_span("my_span", "custom")
span.set_input({"query": "What is AI?"})
span.set_output({"answer": "AI stands for Artificial Intelligence."})
span.finish()

cozeloop.close()
```
```

--------------------------------

### Set Environment Variables for CozeLoop SDK

Source: https://github.com/coze-dev/cozeloop-python/blob/main/README.md

Configure your environment variables with your CozeLoop workspace ID, OAuth client ID, private key, and public key ID before initializing the SDK.

```bash
export COZELOOP_WORKSPACE_ID=your workspace id
export COZELOOP_JWT_OAUTH_CLIENT_ID=your client id
export COZELOOP_JWT_OAUTH_PRIVATE_KEY=your private key
export COZELOOP_JWT_OAUTH_PUBLIC_KEY_ID=your public key id
```

--------------------------------

### Span Interface - Context Manager Usage

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Use `with cozeloop.start_span(...)` for automatic span finishing, even if exceptions occur.

```python
# Context manager usage — finish() called automatically on __exit__
with cozeloop.start_span("guarded_op", "custom") as s:
    s.set_input("safe input")
    # span.finish() is called even if an exception is raised
```

--------------------------------

### Execute a Prompt (Sync, Non-streaming)

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Executes a managed prompt on the CozeLoop backend synchronously without streaming the response. Useful for obtaining a complete response at once.

```python
import asyncio
import cozeloop
from cozeloop.entities.prompt import Message, Role, ExecuteResult

client = cozeloop.new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
)

# --- Sync, non-streaming ---
result: ExecuteResult = client.execute_prompt(
    prompt_key="ptaas_demo",
    version="0.0.1",
    variable_vals={
        "topic":        "artificial intelligence",
        "user_request": "explain what is machine learning",
    },
    messages=[Message(role=Role.USER, content="Keep the answer brief.")],
    stream=False,
    timeout=60,  # seconds; default 600
)
print(result.message)       # Message object with role + content
print(result.finish_reason) # "stop", "length", etc.
print(result.usage)         # TokenUsage(input_tokens=..., output_tokens=...)
```

--------------------------------

### openai_wrapper

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Wraps an OpenAI client (sync or async) to automatically emit CozeLoop spans for API calls.

```APIDOC
## OpenAI SDK Integration

### `openai_wrapper(client)` — Wrap an OpenAI client for automatic tracing

Patches `client.chat.completions.create` (and `client.responses.create` if available) to emit `model`-type spans automatically, capturing inputs, outputs, token usage, model name, and call options. Works with both `OpenAI` and `AsyncOpenAI` (and their Azure variants).

### Parameters
- `client` (`openai.OpenAI` or `openai.AsyncOpenAI`): The OpenAI client instance to wrap.

### Request Example (Sync OpenAI client)
```python
import os
import asyncio
from openai import OpenAI, AsyncOpenAI
from cozeloop import new_client
from cozeloop.integration.wrapper import openai_wrapper

coze_client = new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
)

# --- Sync OpenAI client ---
oai_client = openai_wrapper(OpenAI(api_key=os.environ["OPENAI_API_KEY"]))

response = oai_client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What is the capital of France?"}],
    temperature=0.5,
    max_tokens=128,
)
print(response.choices[0].message.content)

# --- Async OpenAI client with streaming ---
async_oai_client = openai_wrapper(AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"]))

async def run_stream():
    stream = await async_oai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Count to 5."}
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)
    print()

asyncio.run(run_stream())
coze_client.close()
```
```

--------------------------------

### Execute a prompt against an LLM (asynchronous)

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Executes a managed prompt on the CozeLoop backend asynchronously, filling variables and calling the configured LLM. Supports streaming responses. Returns StreamReader[ExecuteResult] for streaming or ExecuteResult for non-streaming.

```APIDOC
## aexecute_prompt(prompt_key, ...)

### Description
Execute a managed prompt asynchronously against an LLM, with optional streaming.

### Method Signature
`aexecute_prompt(prompt_key: str, ..., stream: bool = False, timeout: Optional[int] = None) -> Union[ExecuteResult, StreamReader[ExecuteResult]]`

### Parameters
- **prompt_key** (str) - Required - The key of the prompt to execute.
- **version** (str, optional) - The version of the prompt.
- **label** (str, optional) - The label of the prompt.
- **variable_vals** (Dict[str, Any], optional) - Values for prompt variables.
- **messages** (List[Message], optional) - Additional messages to include in the prompt.
- **stream** (bool) - If True, returns a StreamReader for streaming results. Defaults to False.
- **timeout** (int, optional) - Execution timeout in seconds.

### Returns
- `ExecuteResult` - If stream is False.
- `StreamReader[ExecuteResult]` - If stream is True.

### Example (Non-streaming)
```python
import asyncio
import cozeloop
from cozeloop.entities.prompt import Message, Role, ExecuteResult

client = cozeloop.new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
)

async def run_sync():
    result: ExecuteResult = await client.aexecute_prompt(
        prompt_key="ptaas_demo",
        version="0.0.1",
        variable_vals={
            "topic":        "artificial intelligence",
            "user_request": "explain what is machine learning",
        },
        messages=[Message(role=Role.USER, content="Keep the answer brief.")],
        stream=False,
        timeout=60,
    )
    print(result.message)
    print(result.finish_reason)
    print(result.usage)

asyncio.run(run_sync())
client.close()
```

### Example (Streaming)
```python
import asyncio
import cozeloop
from cozeloop.entities.prompt import Message, Role

client = cozeloop.new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
)

async def run_stream():
    stream_reader = await client.aexecute_prompt(
        prompt_key="ptaas_demo",
        version="0.0.1",
        variable_vals={
            "topic":        "artificial intelligence",
            "user_request": "explain what is machine learning",
        },
        messages=[Message(role=Role.USER, content="Keep the answer brief.")],
        stream=True,
        timeout=60,
    )
    async for result in stream_reader:
        print(result.message.content, end="", flush=True)

asyncio.run(run_stream())
client.close()
```
```

--------------------------------

### Retrieve a Managed Prompt from Prompt Hub

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Fetches a Prompt object from the CozeLoop Prompt Hub using the client. Supports retrieving the latest version, a specific version, or a version by label. Results are cached.

```python
import cozeloop

client = cozeloop.new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
    prompt_trace=True,   # emit a span for this get_prompt call
)

# Fetch the latest version of a prompt
prompt = client.get_prompt(prompt_key="my_assistant_prompt")

# Fetch a specific version
prompt_v1 = client.get_prompt(prompt_key="my_assistant_prompt", version="0.0.1")

# Fetch by label
prompt_prod = client.get_prompt(prompt_key="my_assistant_prompt", label="production")

if prompt:
    print(f"key={prompt.prompt_key}  version={prompt.version}")
    if prompt.prompt_template:
        for msg in prompt.prompt_template.messages:
            print(f"  [{msg.role}] {msg.content}")
    if prompt.llm_config:
        print(f"  temperature={prompt.llm_config.temperature}")
    if prompt.tools:
        print(f"  tools={[t.function.name for t in prompt.tools if t.function]}")

client.close()
```

--------------------------------

### Wrap OpenAI Client for Automatic Tracing

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Automatically traces OpenAI client calls by patching `chat.completions.create` and `responses.create`. This wrapper captures inputs, outputs, token usage, model name, and call options for both synchronous and asynchronous OpenAI clients, including Azure variants.

```python
import os
import asyncio
from openai import OpenAI, AsyncOpenAI
from cozeloop import new_client
from cozeloop.integration.wrapper import openai_wrapper

coze_client = new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
)

# --- Sync OpenAI client ---
oai_client = openai_wrapper(OpenAI(api_key=os.environ["OPENAI_API_KEY"]))

response = oai_client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What is the capital of France?"}],
    temperature=0.5,
    max_tokens=128,
)
print(response.choices[0].message.content)

# --- Async OpenAI client with streaming ---
async_oai_client = openai_wrapper(AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"]))

async def run_stream():
    stream = await async_oai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Count to 5."}],
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)
    print()

asyncio.run(run_stream())
coze_client.close()
```

--------------------------------

### LangChain Callback Handler for CozeLoop Tracing

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Integrates CozeLoop tracing with LangChain by providing a callback handler. This handler automatically generates CozeLoop spans for LangChain nodes like LLM calls, prompts, chains, tools, and graphs. Customize span names and tags using provided functions.

```python
import logging
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from cozeloop import new_client, set_log_level
from cozeloop.integration.langchain.trace_callback import LoopTracer

set_log_level(logging.DEBUG)

client = new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
)

# Optional: customize span names and add tags per node
def rename_node(node_name: str) -> str:
    return f"[app] {node_name}"

def node_tags(node_name: str) -> dict:
    return {"app_version": "2.1.0", "node": node_name}

# Get a fresh callback handler for this request
tracer = LoopTracer.get_callback_handler(
    client=client,
    modify_name_fn=rename_node,
    add_tags_fn=node_tags,
    tags={"env": "production"},
    # child_of=parent_span,       # attach to an existing parent span
    # state_span_ctx_key="span",  # for LangGraph state-based context propagation
)

llm = ChatOpenAI(model="gpt-4o")
chain = llm | StrOutputParser()

output = chain.invoke(
    "Explain the attention mechanism in transformers.",
    config=RunnableConfig(callbacks=[tracer])
)
print(output)

# Access the trace_id for this request
print("trace_id:", tracer.trace_id)

client.close()
```

--------------------------------

### Retrieve a managed prompt

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Fetches a Prompt object from the CozeLoop Prompt Hub using a local cache. Results are cached to avoid repeated network calls. Returns None if not found.

```APIDOC
## get_prompt(prompt_key, version, label)

### Description
Retrieve a managed prompt from the CozeLoop Prompt Hub.

### Method Signature
`get_prompt(prompt_key: str, version: Optional[str] = None, label: Optional[str] = None)`

### Parameters
- **prompt_key** (str) - Required - The unique key of the prompt.
- **version** (str, optional) - The specific version of the prompt to retrieve.
- **label** (str, optional) - The label (e.g., 'production') of the prompt to retrieve.

### Returns
- `Prompt` object if found, otherwise `None`.

### Example
```python
import cozeloop

client = cozeloop.new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
    prompt_trace=True,
)

# Fetch the latest version of a prompt
prompt = client.get_prompt(prompt_key="my_assistant_prompt")

# Fetch a specific version
prompt_v1 = client.get_prompt(prompt_key="my_assistant_prompt", version="0.0.1")

# Fetch by label
prompt_prod = client.get_prompt(prompt_key="my_assistant_prompt", label="production")

client.close()
```
```

--------------------------------

### Configure CozeLoop SDK Logging

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Manages the logging level and handlers for the CozeLoop SDK. The default log level is WARNING, outputting to stdout. Adjust the level to DEBUG during development for more detailed logs.

```python
import logging
import cozeloop
from cozeloop import set_log_level, add_log_handler

# Raise to DEBUG during development
set_log_level(logging.DEBUG)
```

--------------------------------

### Fill prompt template variables

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Replaces template variables and placeholder variables in a Prompt with provided values. Returns a List[Message] ready to pass to an LLM.

```APIDOC
## prompt_format(prompt, variables)

### Description
Fill prompt template variables and placeholder variables.

### Method Signature
`prompt_format(prompt: Prompt, variables: Dict[str, Any]) -> List[Message]`

### Parameters
- **prompt** (Prompt) - Required - The Prompt object to format.
- **variables** (Dict[str, Any]) - Required - A dictionary containing variable names and their corresponding values.

### Returns
- `List[Message]` - A list of Message objects ready for LLM input.

### Example
```python
import cozeloop
from cozeloop.entities.prompt import Message, Role

client = cozeloop.new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
)

prompt = client.get_prompt(prompt_key="prompt_hub_demo", version="0.0.1")

if prompt:
    formatted = client.prompt_format(prompt, {
        "topic":    "machine learning",
        "question": "What is overfitting?",
        "placeholder1": [
            Message(role=Role.USER,      content="Hi!"),
            Message(role=Role.ASSISTANT, content="Hello! How can I help?"),
        ],
    })

client.close()
```
```

--------------------------------

### Fill Prompt Template Variables

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Replaces template variables (e.g., `{{var1}}`) and placeholder variables in a Prompt object with provided values. Returns a List[Message] ready for LLM input.

```python
import json
import cozeloop
from cozeloop.entities.prompt import Message, Role

client = cozeloop.new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
)

# Prompt template stored on platform:
#   System:      "You are a helpful bot. Topic: {{topic}}"
#   Placeholder: placeholder1
#   User:        "My question is {{question}}"

prompt = client.get_prompt(prompt_key="prompt_hub_demo", version="0.0.1")

if prompt:
    formatted = client.prompt_format(prompt, {
        "topic":    "machine learning",       # string variable
        "question": "What is overfitting?",   # string variable
        "placeholder1": [
            Message(role=Role.USER,      content="Hi!"),
            Message(role=Role.ASSISTANT, content="Hello! How can I help?"),
        ],
    })

    print(json.dumps(
        [m.model_dump(exclude_none=True) for m in formatted],
        ensure_ascii=False, indent=2
    ))
    # Expected output:
    # [
    #   {"role": "system",    "content": "You are a helpful bot. Topic: machine learning"},
    #   {"role": "user",      "content": "Hi!"},
    #   {"role": "assistant", "content": "Hello! How can I help?"},
    #   {"role": "user",      "content": "My question is What is overfitting?"}
    # ]

client.close()
```

--------------------------------

### Forcing Span Reporting

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Explains how to use `cozeloop.flush()` to immediately send all queued spans, typically used before application exit.

```APIDOC
### `flush()` — Force-report all queued spans

Blocks until all spans in the async upload queue have been sent to the CozeLoop server. Use before process exit when `close()` is not suitable.

```python
import cozeloop

span = cozeloop.start_span("batch_job", "custom")
span.set_output("done")
span.finish()

# Force flush without closing the client (client remains usable afterwards)
cozeloop.flush()
```
```

--------------------------------

### set_log_level / add_log_handler

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Configure the logging level and handlers for the CozeLoop SDK.

```APIDOC
## Logging Configuration

### `set_log_level(level)` / `add_log_handler(handler)` — Configure SDK logging

The SDK uses the `cozeloop` logger. Default level is `WARNING` with output to stdout.

### Parameters
- `level` (logging level): The minimum severity level to log (e.g., `logging.DEBUG`, `logging.INFO`).
- `handler` (logging.Handler): A logging handler to add to the SDK's logger.

### Request Example
```python
import logging
import cozeloop
from cozeloop import set_log_level, add_log_handler

# Raise to DEBUG during development
set_log_level(logging.DEBUG)

# Add a file handler for example
# file_handler = logging.FileHandler("cozeloop.log")
# add_log_handler(file_handler)
```
```

--------------------------------

### Span Interface - Cross-Process Context Propagation

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Generate headers for context propagation and reconstruct parent context from incoming headers to create child spans.

```python
# Cross-process / cross-service context propagation
headers = span.to_header()
# {"X-Cozeloop-Traceparent": "...", "X-Cozeloop-Tracestate": "..."}

# On the receiving service, reconstruct parent context from incoming headers
parent_ctx = cozeloop.get_span_from_header(headers)
child = cozeloop.start_span("downstream_op", "custom", child_of=parent_ctx)
child.finish()
```

--------------------------------

### Replace Default Logging Handler

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Replaces the default stdout logging handler with a file handler to log messages to 'cozeloop.log'. Ensure the `add_log_handler` function is available in your scope.

```python
file_handler = logging.FileHandler("cozeloop.log")
file_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
add_log_handler(file_handler)

span = cozeloop.start_span("test", "custom")
span.finish()
cozeloop.close()
```

--------------------------------

### Span Interface - Core Setters

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Demonstrates core setters for input, output, and error on a span. Use `set_status_code(-1)` to indicate an error.

```python
span = cozeloop.start_span("my_op", "custom")

# Core input / output / error
span.set_input({"messages": [{"role": "user", "content": "Hello"}]})
span.set_output("Hi there!")
span.set_error(ValueError("something went wrong"))
span.set_status_code(-1)          # Non-zero = error
```

--------------------------------

### LoopTracer.get_callback_handler

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Provides a LangChain callback handler that automatically creates and reports CozeLoop spans for LangChain nodes.

```APIDOC
## LangChain / LangGraph Integration

### `LoopTracer.get_callback_handler(...)` — LangChain callback handler

Returns a `LoopTraceCallbackHandler` (a `BaseCallbackHandler`) that automatically creates and reports CozeLoop spans for every LangChain node: LLM calls (`model` spans), prompt templates (`prompt` spans), chain steps (`chain` spans), tools (`tool` spans), and LangGraph graphs (`graph` spans). Get a fresh handler per request.

### Parameters
- `client` (`cozeloop.Client`): The CozeLoop client instance.
- `modify_name_fn` (callable, optional): A function to modify the name of the spans.
- `add_tags_fn` (callable, optional): A function to add tags to the spans.
- `tags` (dict, optional): Additional tags to apply to all spans created by this handler.
- `child_of` (Span, optional): Attach the created spans to an existing parent span.
- `state_span_ctx_key` (str, optional): Key for LangGraph state-based context propagation.

### Request Example
```python
import logging
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from cozeloop import new_client, set_log_level
from cozeloop.integration.langchain.trace_callback import LoopTracer

set_log_level(logging.DEBUG)

client = new_client(
    workspace_id="7421xxxxxxxxxxxxxxxx",
    api_token="pat_xxxxxxxxxxxxxxxx",
)

# Optional: customize span names and add tags per node
def rename_node(node_name: str) -> str:
    return f"[app] {node_name}"

def node_tags(node_name: str) -> dict:
    return {"app_version": "2.1.0", "node": node_name}

# Get a fresh callback handler for this request
tracer = LoopTracer.get_callback_handler(
    client=client,
    modify_name_fn=rename_node,
    add_tags_fn=node_tags,
    tags={"env": "production"},
    # child_of=parent_span,       # attach to an existing parent span
    # state_span_ctx_key="span",  # for LangGraph state-based context propagation
)

llm = ChatOpenAI(model="gpt-4o")
chain = llm | StrOutputParser()

output = chain.invoke(
    "Explain the attention mechanism in transformers.",
    config=RunnableConfig(callbacks=[tracer])
)
print(output)

# Access the trace_id for this request
print("trace_id:", tracer.trace_id)

client.close()
```
```

--------------------------------

### Automatic Span Creation with `@observe` Decorator

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Use the `@observe()` decorator to automatically create spans for function calls, capturing arguments and return values. Exceptions are also recorded.

```python
import asyncio
import cozeloop
from cozeloop.decorator import observe

# Basic usage — span name defaults to the function name
@observe()
def preprocess(text: str) -> str:
    return text.strip().lower()

# With explicit name, span type, and static tags
@observe(name="llm_inference", span_type="model", tags={"model": "gpt-4o"})
def call_llm(prompt: str) -> str:
    return "mocked LLM response"

# Async function — works identically
@observe()
async def async_fetch(url: str) -> dict:
    await asyncio.sleep(0.1)
    return {"url": url, "status": 200}

# Generator function — output collected into a list after full consumption
@observe()
def stream_tokens(text: str):
    for token in text.split():
        yield token

# Custom input/output processors to scrub or reshape traced data
@observe(
    process_inputs=lambda inp: {"query": inp["kwargs"].get("query", "")},
    process_outputs=lambda out: {"summary": str(out)[:200]},
)
def search_and_summarize(query: str) -> str:
    return f"Results for: {query}"
```

--------------------------------

### `@observe` Decorator - Automatic Span Creation

Source: https://context7.com/coze-dev/cozeloop-python/llms.txt

Details the `@observe` decorator for automatically creating spans around function calls, capturing inputs, outputs, and exceptions.

```APIDOC
## `@observe` Decorator — Automatic span creation for functions

### `cozeloop.decorator.observe` — Decorator for tracing any function

Wraps a function (sync, async, generator, or async generator) so that a span is automatically created on each call, capturing function arguments as `input` and the return value as `output`. Exceptions are recorded on the span automatically.

```python
import asyncio
import cozeloop
from cozeloop.decorator import observe

# Basic usage — span name defaults to the function name
@observe()
def preprocess(text: str) -> str:
    return text.strip().lower()

# With explicit name, span type, and static tags
@observe(name="llm_inference", span_type="model", tags={"model": "gpt-4o"})
def call_llm(prompt: str) -> str:
    return "mocked LLM response"

# Async function — works identically
@observe()
async def async_fetch(url: str) -> dict:
    await asyncio.sleep(0.1)
    return {"url": url, "status": 200}

# Generator function — output collected into a list after full consumption
@observe()
def stream_tokens(text: str):
    for token in text.split():
        yield token

# Custom input/output processors to scrub or reshape traced data
@observe(
    process_inputs=lambda inp: {"query": inp["kwargs"].get("query", "")},
    process_outputs=lambda out: {"summary": str(out)[:200]},
)
def search_and_summarize(query: str) -> str:
    return f"Results for: {query}"
```
```