Anthropic Task Planner

pip install mantisdk anthropic python-dotenv

# .env file
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-key
INSIGHT_SECRET_KEY=sk-lf-your-key
ANTHROPIC_API_KEY=sk-ant-your-anthropic-key

#!/usr/bin/env python3
"""Anthropic Task Planner Agent with Mantis SDK Tracing."""
 
import argparse
import asyncio
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())
    completed_at: Optional[str] = None
 
 
class TaskStore:
    """Simple in-memory task store."""
 
    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 get_task(self, task_id: str) -> Optional[Task]:
        return self.tasks.get(task_id)
 
    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"
            task.completed_at = datetime.now().isoformat()
        return task
 
 
task_store = TaskStore()
 
 
# Tool implementations
def create_task(title: str, description: str, priority: str = "medium") -> dict:
    if priority not in ("low", "medium", "high"):
        return {"status": "error", "error_message": f"Invalid priority: {priority}"}
    task = task_store.create_task(title, description, priority)
    return {
        "status": "success",
        "task": {"id": task.id, "title": task.title, "priority": task.priority}
    }
 
 
def list_tasks(status: Optional[str] = None) -> dict:
    if status and status not in ("pending", "in_progress", "completed"):
        return {"status": "error", "error_message": f"Invalid status: {status}"}
    tasks = task_store.list_tasks(status)
    return {
        "status": "success",
        "tasks": [{"id": t.id, "title": t.title, "status": t.status} for t in tasks],
        "count": len(tasks)
    }
 
 
def get_task_details(task_id: str) -> dict:
    task = task_store.get_task(task_id)
    if not task:
        return {"status": "error", "error_message": f"Task '{task_id}' not found"}
    return {
        "status": "success",
        "task": {
            "id": task.id, "title": task.title, "description": task.description,
            "status": task.status, "priority": task.priority
        }
    }
 
 
def complete_task(task_id: str) -> dict:
    task = task_store.complete_task(task_id)
    if not task:
        return {"status": "error", "error_message": f"Task '{task_id}' not found"}
    return {
        "status": "success",
        "message": f"Task '{task.title}' marked as completed"
    }
 
 
TOOL_FUNCTIONS = {
    "create_task": create_task,
    "list_tasks": list_tasks,
    "get_task_details": get_task_details,
    "complete_task": complete_task,
}
 
TOOLS = [
    {
        "name": "create_task",
        "description": "Create a new task with title, description, and priority",
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "The task title"},
                "description": {"type": "string", "description": "Task description"},
                "priority": {
                    "type": "string",
                    "enum": ["low", "medium", "high"],
                    "description": "Task priority"
                }
            },
            "required": ["title", "description"]
        }
    },
    {
        "name": "list_tasks",
        "description": "List all tasks, optionally filtered by status",
        "input_schema": {
            "type": "object",
            "properties": {
                "status": {
                    "type": "string",
                    "enum": ["pending", "in_progress", "completed"]
                }
            }
        }
    },
    {
        "name": "get_task_details",
        "description": "Get detailed information about a specific task",
        "input_schema": {
            "type": "object",
            "properties": {"task_id": {"type": "string"}},
            "required": ["task_id"]
        }
    },
    {
        "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 planning agent with tool use."""
 
    def __init__(self, model: str = "claude-sonnet-4-20250514"):
        self.client = anthropic.Anthropic()
        self.model = model
        self.messages = []
        self.system_prompt = """You are a helpful task planning assistant.
Help users manage their tasks effectively using the available tools."""
 
    def reset(self):
        self.messages = []
 
    async def chat(self, user_message: str) -> str:
        """Process a user message - auto-instrumented by Anthropic instrumentor."""
        self.messages.append({"role": "user", "content": user_message})
 
        # This API call is auto-instrumented
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=self.system_prompt,
            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 use in the response."""
        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 API call - also auto-instrumented
        final = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=self.system_prompt,
            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 run_interactive_session(agent: TaskPlannerAgent):
    """Run an interactive CLI session."""
    print("\n" + "="*60)
    print("  Anthropic Task Planner (with Mantis Tracing)")
    print("="*60)
    print("\nI can help you manage your tasks:")
    print("  - Create new tasks with priorities")
    print("  - List and filter tasks by status")
    print("  - Mark tasks as complete")
    print("\nType 'quit' to exit.")
    print("="*60 + "\n")
 
    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n\nGoodbye!")
            break
 
        if not user_input:
            continue
        if user_input.lower() in ("quit", "exit", "q"):
            print("\nGoodbye!")
            break
 
        print("\nAssistant: ", end="", flush=True)
        try:
            response = await agent.chat(user_input)
            print(response)
        except Exception as e:
            print(f"\n[Error] {e}")
        print()
 
 
def main():
    parser = argparse.ArgumentParser(description="Anthropic Task Planner")
    parser.add_argument("--model", default="claude-sonnet-4-20250514")
    parser.add_argument("--query", "-q", help="Single query")
    parser.add_argument("--trace-to-console", action="store_true")
    parser.add_argument("--no-tracing", action="store_true")
    args = parser.parse_args()
 
    if not args.no_tracing:
        setup_tracing(trace_to_console=args.trace_to_console)
 
    agent = TaskPlannerAgent(model=args.model)
 
    try:
        if args.query:
            asyncio.run(agent.chat(args.query).then(print))
        else:
            asyncio.run(run_interactive_session(agent))
    finally:
        if not args.no_tracing:
            tracing.shutdown()
 
 
if __name__ == "__main__":
    main()

python task_planner_agent.py

You: Create a task to review the Q4 report
Assistant: I've created a new task for you...

You: List all my tasks
Assistant: Here are your current tasks...

You: Mark that task as complete
Assistant: Done! The task has been marked as completed.

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

# No manual tracing needed - auto-instrumented
response = self.client.messages.create(
    model=self.model,
    max_tokens=1024,
    messages=self.messages,
    tools=TOOLS,
)

if response.stop_reason == "tool_use":
    return await self._handle_tool_use(response)