Anthropic Integration

pip install mantisdk anthropic

from dotenv import load_dotenv
load_dotenv()
 
# Initialize tracing FIRST
import mantisdk.tracing as tracing
 
tracing.init(service_name="anthropic-agent", instrument_default=False)
tracing.instrument(names=["anthropic"])
 
# NOW import Anthropic
import anthropic
 
# Create client and use - automatically traced
client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello, Claude!"}]
)
print(response.content[0].text)
 
# Clean up
tracing.shutdown()

import anthropic
import mantisdk.tracing as tracing
 
client = anthropic.Anthropic()
 
tools = [
    {
        "name": "get_weather",
        "description": "Get weather for a city",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string"}
            },
            "required": ["city"]
        }
    }
]
 
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "What's the weather in Tokyo?"}]
)
# Tool calls are automatically traced

#!/usr/bin/env python3
"""Anthropic Task Planner Agent with Mantis SDK Tracing."""
 
import json
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional
from dotenv import load_dotenv
load_dotenv()
 
# Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured
 
 
def setup_tracing(trace_to_console: bool = False):
    """Initialize mantisdk tracing with Anthropic instrumentation."""
    exporters = []
 
    if is_insight_configured():
        exporters.append(tracing.insight_exporter())
        print("[Tracing] Insight exporter configured")
    else:
        print("[Tracing] No Insight credentials found")
 
    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())
 
    tracing.init(
        service_name="anthropic-task-planner",
        exporters=exporters if exporters else None,
        instrument_default=False,
    )
 
    tracing.instrument(names=["anthropic"])
    print("[Tracing] Anthropic instrumentation enabled")
 
 
# NOW import Anthropic
import anthropic
 
 
# Task data model
@dataclass
class Task:
    id: str
    title: str
    description: str
    status: str = "pending"
    priority: str = "medium"
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
 
 
# In-memory task store
class TaskStore:
    def __init__(self):
        self.tasks: Dict[str, Task] = {}
 
    def create_task(self, title: str, description: str, priority: str = "medium") -> Task:
        import uuid
        task_id = str(uuid.uuid4())[:8]
        task = Task(id=task_id, title=title, description=description, priority=priority)
        self.tasks[task_id] = task
        return task
 
    def list_tasks(self, status: Optional[str] = None) -> List[Task]:
        tasks = list(self.tasks.values())
        if status:
            tasks = [t for t in tasks if t.status == status]
        return tasks
 
    def complete_task(self, task_id: str) -> Optional[Task]:
        task = self.tasks.get(task_id)
        if task:
            task.status = "completed"
        return task
 
 
task_store = TaskStore()
 
 
# Tool implementations
def create_task(title: str, description: str, priority: str = "medium") -> dict:
    task = task_store.create_task(title, description, priority)
    return {"status": "success", "task": {"id": task.id, "title": task.title}}
 
 
def list_tasks(status: Optional[str] = None) -> dict:
    tasks = task_store.list_tasks(status)
    return {"status": "success", "tasks": [{"id": t.id, "title": t.title} for t in tasks]}
 
 
def complete_task(task_id: str) -> dict:
    task = task_store.complete_task(task_id)
    if task:
        return {"status": "success", "message": f"Completed: {task.title}"}
    return {"status": "error", "error_message": f"Task {task_id} not found"}
 
 
TOOL_FUNCTIONS = {
    "create_task": create_task,
    "list_tasks": list_tasks,
    "complete_task": complete_task,
}
 
TOOLS = [
    {
        "name": "create_task",
        "description": "Create a new task",
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "description": {"type": "string"},
                "priority": {"type": "string", "enum": ["low", "medium", "high"]}
            },
            "required": ["title", "description"]
        }
    },
    {
        "name": "list_tasks",
        "description": "List all tasks",
        "input_schema": {
            "type": "object",
            "properties": {
                "status": {"type": "string", "enum": ["pending", "completed"]}
            }
        }
    },
    {
        "name": "complete_task",
        "description": "Mark a task as completed",
        "input_schema": {
            "type": "object",
            "properties": {"task_id": {"type": "string"}},
            "required": ["task_id"]
        }
    }
]
 
 
class TaskPlannerAgent:
    """Anthropic-based task planner with tool use."""
 
    def __init__(self, model: str = "claude-sonnet-4-20250514"):
        self.client = anthropic.Anthropic()
        self.model = model
        self.messages = []
        self.system = "You are a task planning assistant. Help users manage tasks."
 
    async def chat(self, user_message: str) -> str:
        """Process a user message - auto-instrumented."""
        self.messages.append({"role": "user", "content": user_message})
 
        # This call is auto-instrumented
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=self.system,
            messages=self.messages,
            tools=TOOLS,
        )
 
        if response.stop_reason == "tool_use":
            return await self._handle_tool_use(response)
 
        text = "".join(b.text for b in response.content if hasattr(b, "text"))
        self.messages.append({"role": "assistant", "content": response.content})
        return text
 
    async def _handle_tool_use(self, response) -> str:
        """Handle tool calls."""
        self.messages.append({"role": "assistant", "content": response.content})
 
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                func = TOOL_FUNCTIONS.get(block.name)
                if func:
                    result = func(**block.input)
                else:
                    result = {"status": "error", "error_message": "Unknown tool"}
 
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result)
                })
 
        self.messages.append({"role": "user", "content": tool_results})
 
        # Follow-up call - also auto-instrumented
        final = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=self.system,
            messages=self.messages,
            tools=TOOLS,
        )
 
        if final.stop_reason == "tool_use":
            return await self._handle_tool_use(final)
 
        text = "".join(b.text for b in final.content if hasattr(b, "text"))
        self.messages.append({"role": "assistant", "content": final.content})
        return text
 
 
async def main():
    import asyncio
 
    setup_tracing(trace_to_console=False)
 
    agent = TaskPlannerAgent()
 
    # Example conversation
    queries = [
        "Create a task to review the Q4 report",
        "List all my tasks",
    ]
 
    for query in queries:
        print(f"\nUser: {query}")
        response = await agent.chat(query)
        print(f"Assistant: {response}")
 
    tracing.shutdown()
 
 
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

chat (function call)
├── anthropic.messages.create (auto-instrumented)
│   ├── model: claude-sonnet-4-20250514
│   ├── input_tokens: 180
│   ├── output_tokens: 45
│   └── tool_use: create_task
└── anthropic.messages.create (follow-up)
    ├── model: claude-sonnet-4-20250514
    ├── input_tokens: 250
    └── output_tokens: 30

import anthropic
import mantisdk.tracing as tracing
 
client = anthropic.Anthropic()
 
def analyze_text(text: str) -> dict:
    with tracing.trace("analyze_text", input=text) as span:
        span.set_attribute(tracing.semconv.USER_ID, "user-123")
 
        # Preprocess
        with tracing.span("preprocess"):
            cleaned = preprocess(text)
 
        # LLM call - auto-instrumented as child span
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": cleaned}]
        )
 
        result = response.content[0].text
        span.set_attribute(tracing.semconv.TRACE_OUTPUT, result)
        return {"analysis": result}