Claude Agent SDK

1
from openinference.instrumentation.claude_agent_sdk import ClaudeAgentSDKInstrumentor
2
from respan import Respan
3
from respan_instrumentation_openinference import OpenInferenceInstrumentor
4
5
respan = Respan(
6
    instrumentations=[OpenInferenceInstrumentor(ClaudeAgentSDKInstrumentor)],
7
    customer_identifier="user_123",
8
    thread_identifier="conv_001",
9
    metadata={"service": "claude-agent", "plan": "pro"},
10
)

1
import os
2
3
import claude_agent_sdk
4
from claude_agent_sdk.types import ClaudeAgentOptions
5
from respan import propagate_attributes
6
7
8
def extract_text(message: object) -> str:
9
    parts = []
10
    for block in getattr(message, "content", []) or []:
11
        text = getattr(block, "text", None)
12
        if isinstance(text, str) and text:
13
            parts.append(text)
14
    return "\n".join(parts)
15
16
17
async def handle_request(user_id: str, prompt: str):
18
    options = ClaudeAgentOptions(
19
        model="sonnet",
20
        max_turns=1,
21
        env={"ANTHROPIC_API_KEY": os.environ["ANTHROPIC_API_KEY"]},
22
    )
23
24
    with propagate_attributes(
25
        customer_identifier=user_id,
26
        thread_identifier="conv_abc_123",
27
        metadata={"plan": "pro"},
28
    ):
29
        assistant_text = ""
30
        async for message in claude_agent_sdk.query(prompt=prompt, options=options):
31
            if type(message).__name__ == "AssistantMessage":
32
                assistant_text = extract_text(message)
33
34
        print(assistant_text)

1
import asyncio
2
import os
3
4
import claude_agent_sdk
5
from claude_agent_sdk.types import ClaudeAgentOptions
6
from openinference.instrumentation.claude_agent_sdk import ClaudeAgentSDKInstrumentor
7
from respan import Respan, task, workflow
8
from respan_instrumentation_openinference import OpenInferenceInstrumentor
9
10
respan = Respan(
11
    instrumentations=[OpenInferenceInstrumentor(ClaudeAgentSDKInstrumentor)]
12
)
13
14
15
def extract_text(message: object) -> str:
16
    parts = []
17
    for block in getattr(message, "content", []) or []:
18
        text = getattr(block, "text", None)
19
        if isinstance(text, str) and text:
20
            parts.append(text)
21
    return "\n".join(parts)
22
23
24
@task(name="draft_answer")
25
async def draft_answer(prompt: str) -> str:
26
    options = ClaudeAgentOptions(
27
        model="sonnet",
28
        max_turns=1,
29
        env={"ANTHROPIC_API_KEY": os.environ["ANTHROPIC_API_KEY"]},
30
    )
31
32
    assistant_text = ""
33
    async for message in claude_agent_sdk.query(prompt=prompt, options=options):
34
        if type(message).__name__ == "AssistantMessage":
35
            assistant_text = extract_text(message)
36
37
    return assistant_text
38
39
40
@workflow(name="customer_support_flow")
41
async def handle_ticket():
42
    summary = await draft_answer("Summarize a billing issue in one sentence.")
43
    print(summary)
44
45
46
asyncio.run(handle_ticket())
47
respan.flush()

1
import asyncio
2
import os
3
4
import claude_agent_sdk
5
from claude_agent_sdk.types import ClaudeAgentOptions
6
7
8
def extract_text(message: object) -> str:
9
    parts = []
10
    for block in getattr(message, "content", []) or []:
11
        text = getattr(block, "text", None)
12
        if isinstance(text, str) and text:
13
            parts.append(text)
14
    return "\n".join(parts)
15
16
17
async def main():
18
    options = ClaudeAgentOptions(
19
        model="sonnet",
20
        max_turns=1,
21
        env={"ANTHROPIC_API_KEY": os.environ["ANTHROPIC_API_KEY"]},
22
    )
23
24
    async for message in claude_agent_sdk.query(
25
        prompt="Say hello in three languages.",
26
        options=options,
27
    ):
28
        if type(message).__name__ == "AssistantMessage":
29
            print(extract_text(message))
30
31
32
asyncio.run(main())

1
import asyncio
2
import os
3
4
import claude_agent_sdk
5
from claude_agent_sdk.types import ClaudeAgentOptions
6
7
8
async def main():
9
    options = ClaudeAgentOptions(
10
        model="sonnet",
11
        max_turns=1,
12
        env={"ANTHROPIC_API_KEY": os.environ["ANTHROPIC_API_KEY"]},
13
    )
14
15
    message_types = []
16
    async for message in claude_agent_sdk.query(
17
        prompt="Explain recursion in one short paragraph.",
18
        options=options,
19
    ):
20
        message_types.append(type(message).__name__)
21
22
    print(" -> ".join(message_types))
23
24
25
asyncio.run(main())