Claude Code SDK Integration

#!/usr/bin/env python3
import asyncio
from typing import Any
from dotenv import load_dotenv
load_dotenv()
 
# Initialize mantisdk tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured
 
exporters = []
if is_insight_configured():
    exporters.append(tracing.insight_exporter())
 
tracing.init(
    service_name="claude-agent",
    exporters=exporters if exporters else None,
    instrument_default=False,
)
 
# Instrument Claude Agent SDK
tracing.instrument(names=["claude_agent_sdk"])
 
# NOW import the SDK
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    tool,
    create_sdk_mcp_server,
)
 
# Define a tool
@tool(
    "get_weather",
    "Gets the current weather for a given city",
    {"city": str},
)
async def get_weather(args: dict[str, Any]) -> dict[str, Any]:
    city = args["city"]
    # Simulated weather lookup
    weather = "Sunny, 72F"
    return {
        "content": [{"type": "text", "text": f"Weather in {city}: {weather}"}]
    }
 
async def main():
    # Create MCP server with tools
    weather_server = create_sdk_mcp_server(
        name="weather",
        version="1.0.0",
        tools=[get_weather],
    )
 
    # Configure agent options
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-20250514",
        system_prompt="You are a helpful assistant with weather information.",
        mcp_servers={"weather": weather_server},
        allowed_tools=["mcp__weather__get_weather"],
    )
 
    # Use the agent - calls are auto-traced
    with tracing.trace("weather-query", input="What's the weather in Paris?") as span:
        async with ClaudeSDKClient(options=options) as client:
            await client.query("What's the weather like in Paris?")
 
            response_text = []
            async for message in client.receive_response():
                if hasattr(message, "content"):
                    for block in message.content:
                        if hasattr(block, "text"):
                            response_text.append(block.text)
 
            result = "".join(response_text)
            span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
            print(result)
 
if __name__ == "__main__":
    try:
        asyncio.run(main())
    finally:
        tracing.shutdown()

async def handle_user_query(user_id: str, session_id: str, query: str):
    with tracing.trace("user-query", input=query) as span:
        # Add user context
        span.set_attribute(tracing.semconv.USER_ID, user_id)
        span.set_attribute(tracing.semconv.SESSION_ID, session_id)
 
        # Run the agent
        async with ClaudeSDKClient(options=options) as client:
            await client.query(query)
 
            async for message in client.receive_response():
                # Process response
                pass
 
        span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)

async def conversation_session(user_id: str):
    session_id = f"session-{uuid.uuid4()}"
 
    with tracing.trace("conversation", metadata={"session": session_id}) as session_span:
        session_span.set_attribute(tracing.semconv.USER_ID, user_id)
        session_span.set_attribute(tracing.semconv.SESSION_ID, session_id)
 
        async with ClaudeSDKClient(options=options) as client:
            # First turn
            with tracing.span("turn-1", input="Hello!"):
                await client.query("Hello!")
                # ... handle response
 
            # Second turn
            with tracing.span("turn-2", input="What can you help me with?"):
                await client.query("What can you help me with?")
                # ... handle response

async def safe_agent_call(query: str):
    with tracing.trace("agent-call", input=query) as span:
        try:
            async with ClaudeSDKClient(options=options) as client:
                await client.query(query)
 
                async for message in client.receive_response():
                    if hasattr(message, "is_error") and message.is_error:
                        span.set_attribute("error", True)
                        span.set_attribute("error.message", str(message))
                        raise Exception(f"Agent error: {message}")
 
                    # Process successful response
 
        except Exception as e:
            span.set_attribute("error", True)
            span.set_attribute("error.type", type(e).__name__)
            span.set_attribute("error.message", str(e))
            raise