Tool Use với Pydantic — Type-safe tools cho Claude

Claude's tool use là tính năng mạnh, nhưng trong production bạn không thể tin tưởng mù quáng vào output. Điều gì xảy ra nếu Claude trả về author_id là string thay vì integer? Hay thiếu required field? Pydantic — thư viện validation phổ biến nhất trong Python ecosystem — cung cấp một lớp bảo vệ type-safe cho mọi tool call response.

Tại sao cần validate tool call responses?

Khi Claude gọi một tool, nó phải cung cấp JSON arguments theo schema bạn định nghĩa. Trong hầu hết các trường hợp, Claude làm đúng. Nhưng:

• Claude có thể hiểu nhầm schema, đặc biệt với complex nested objects
• Integer fields đôi khi được trả về dưới dạng string
• Optional fields có thể bị bỏ qua hoặc điền sai giá trị
• Trong production với thousands of calls/day, edge cases sẽ xuất hiện

Pydantic validation bắt các lỗi này trước khi chúng propagate vào database hoặc downstream systems — tiết kiệm debugging time và ngăn data corruption.

Setup: Install và import

pip install anthropic pydantic

import anthropic
import json
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime

client = anthropic.Anthropic()

Bước 1: Define Pydantic models

Chúng ta sẽ xây dựng một note-taking system với ba models: Author, Note, và SaveNoteResponse.

class Author(BaseModel):
    """Thong tin tac gia cua note."""
    id: int = Field(..., gt=0, description="Author ID, must be positive integer")
    name: str = Field(..., min_length=1, max_length=100)
    email: Optional[str] = Field(None, pattern=r"^[w.-]+@[w.-]+.w+$")

    @validator('name')
    def name_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError('Name cannot be whitespace only')
        return v.strip()


class Note(BaseModel):
    """Mot ghi chu trong he thong."""
    title: str = Field(..., min_length=3, max_length=200)
    content: str = Field(..., min_length=10)
    author: Author
    tags: List[str] = Field(default_factory=list)
    priority: str = Field(default="normal", pattern=r"^(low|normal|high|urgent)$")
    is_public: bool = Field(default=False)

    @validator('tags')
    def tags_must_be_lowercase(cls, v):
        return [tag.lower().strip() for tag in v if tag.strip()]

    @validator('content')
    def content_must_have_substance(cls, v):
        if len(v.split()) < 3:
            raise ValueError('Content must have at least 3 words')
        return v


class SaveNoteResponse(BaseModel):
    """Response sau khi luu note thanh cong."""
    note_id: str = Field(..., pattern=r"^NOTE-d{6}$")
    saved_at: str  # ISO datetime string
    word_count: int = Field(..., ge=0)
    estimated_read_time: int = Field(..., ge=1, description="Minutes to read")

Bước 2: Define tool từ Pydantic model

Thay vì viết tool schema thủ công, chúng ta generate từ Pydantic models:

def pydantic_to_tool_schema(model_class, tool_name, description):
    """Chuyen Pydantic model thanh Anthropic tool schema."""
    schema = model_class.model_json_schema()

    # Clean up Pydantic-specific metadata
    def clean_schema(s):
        if isinstance(s, dict):
            # Remove 'title' fields (Pydantic adds them, Anthropic doesn't need)
            s.pop('title', None)
            for key, value in s.items():
                if isinstance(value, (dict, list)):
                    clean_schema(value)
        elif isinstance(s, list):
            for item in s:
                clean_schema(item)
        return s

    cleaned = clean_schema(schema)

    return {
        "name": tool_name,
        "description": description,
        "input_schema": cleaned
    }

# Define tools
SAVE_NOTE_TOOL = pydantic_to_tool_schema(
    Note,
    "save_note",
    "Luu mot ghi chu moi vao he thong. Yeu cau title, content, va thong tin author."
)

SEARCH_NOTES_TOOL = {
    "name": "search_notes",
    "description": "Tim kiem notes theo keyword hoac tags",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Keyword tim kiem"},
            "tags": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Filter theo tags"
            },
            "limit": {
                "type": "integer",
                "minimum": 1,
                "maximum": 50,
                "default": 10
            }
        },
        "required": ["query"]
    }
}

Bước 3: Validate tool call input với Pydantic

import random

def handle_save_note(raw_input: dict) -> dict:
    """
    Xu ly save_note tool call voi Pydantic validation.
    raw_input: JSON dict tu Claude's tool call
    Returns: dict response
    """
    # VALIDATE — day la buoc quan trong nhat
    try:
        note = Note(**raw_input)
    except Exception as e:
        # Pydantic validation failed
        return {
            "success": False,
            "error": f"Validation failed: {str(e)}",
            "fields_with_errors": [
                err["loc"][0] for err in e.errors()
            ] if hasattr(e, 'errors') else []
        }

    # Sau khi validate thanh cong, xu ly business logic
    note_id = f"NOTE-{random.randint(100000, 999999)}"
    word_count = len(note.content.split())
    read_time = max(1, word_count // 200)  # ~200 words/minute

    # Build response va validate no cung voi Pydantic
    try:
        response = SaveNoteResponse(
            note_id=note_id,
            saved_at=datetime.now().isoformat(),
            word_count=word_count,
            estimated_read_time=read_time
        )
        return {
            "success": True,
            "data": response.model_dump()
        }
    except Exception as e:
        return {
            "success": False,
            "error": f"Response validation failed: {str(e)}"
        }


def handle_search_notes(raw_input: dict) -> dict:
    """Simulate search — in production would query database."""
    query = raw_input.get("query", "")
    tags = raw_input.get("tags", [])
    limit = raw_input.get("limit", 10)

    # Mock results
    mock_notes = [
        {
            "note_id": f"NOTE-{i:06d}",
            "title": f"Note ve '{query}' so {i}",
            "excerpt": f"Day la excerpt cua note lien quan den {query}...",
            "tags": tags[:2] if tags else ["general"],
            "author_name": "Nguyen Van A"
        }
        for i in range(1, min(limit + 1, 6))
    ]

    return {"results": mock_notes, "total": len(mock_notes)}

Bước 4: Complete flow — từ user request đến validated response

def run_note_taking_agent(user_request: str):
    """
    Complete flow:
    1. User request
    2. Claude decides to use tool
    3. Validate tool call input with Pydantic
    4. Execute tool
    5. Validate response with Pydantic
    6. Return to Claude
    """
    messages = [{"role": "user", "content": user_request}]
    tools = [SAVE_NOTE_TOOL, SEARCH_NOTES_TOOL]

    print(f"User: {user_request}
")

    while True:
        response = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=2048,
            system="""Ban la assistant giup nguoi dung quan ly ghi chu.
            Khi nguoi dung muon luu note, hay dung save_note tool.
            Khi ho muon tim kiem, hay dung search_notes tool.
            Author mac dinh: {id: 1, name: "Default User", email: "user@example.com"}""",
            tools=tools,
            messages=messages
        )

        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, 'text'):
                    print(f"Claude: {block.text}")
            break

        elif response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []

            for block in response.content:
                if block.type == "tool_use":
                    print(f"[Tool Call] {block.name}")
                    print(f"[Input] {json.dumps(block.input, ensure_ascii=False, indent=2)}")

                    # Dispatch to appropriate handler
                    if block.name == "save_note":
                        result = handle_save_note(block.input)
                    elif block.name == "search_notes":
                        result = handle_search_notes(block.input)
                    else:
                        result = {"error": f"Unknown tool: {block.name}"}

                    print(f"[Result] {json.dumps(result, ensure_ascii=False, indent=2)}
")

                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result, ensure_ascii=False),
                        # Nếu validation failed, đánh dấu là error
                        "is_error": not result.get("success", True)
                    })

            messages.append({"role": "user", "content": tool_results})
        else:
            break

Test: Các scenarios validation

# Test 1: Valid note
run_note_taking_agent(
    "Luu note nay: 'Hoc Pydantic' voi noi dung 'Pydantic la thu vien validation manh nhat cho Python, rat phu hop voi FastAPI va Claude tool use.' Tags: python, validation. Priority: high."
)

# Test 2: Invalid data — Claude se nhan error feedback
run_note_taking_agent(
    "Luu note voi title la 'X' (qua ngan) va content la 'Hi' (qua ngan)"
)

# Test 3: Search
run_note_taking_agent(
    "Tim kiem cac notes lien quan den Python"
)

Output mẫu cho valid note:

[Tool Call] save_note
[Input] {
  "title": "Hoc Pydantic",
  "content": "Pydantic la thu vien validation manh nhat...",
  "author": {"id": 1, "name": "Default User", "email": "user@example.com"},
  "tags": ["python", "validation"],
  "priority": "high",
  "is_public": false
}
[Result] {
  "success": true,
  "data": {
    "note_id": "NOTE-847293",
    "saved_at": "2024-12-15T10:30:00.123456",
    "word_count": 18,
    "estimated_read_time": 1
  }
}

Claude: Da luu note "Hoc Pydantic" thanh cong! Note ID: NOTE-847293,
uoc tinh doc trong 1 phut.

Output mẫu cho invalid data:

[Tool Call] save_note
[Input] {"title": "X", "content": "Hi", "author": {...}}
[Result] {
  "success": false,
  "error": "Validation failed: 2 validation errors...",
  "fields_with_errors": ["title", "content"]
}

Claude: Co loi khi luu note: title phai co it nhat 3 ky tu
(hien tai chi co 1), va content phai co it nhat 10 ky tu va 3 tu.
Vui long cung cap thong tin day du hon.

Advanced: Custom validators cho business logic

from pydantic import model_validator

class ExpenseNote(BaseModel):
    """Note cho expense report voi business rules."""
    title: str = Field(..., min_length=3)
    amount: float = Field(..., gt=0)
    currency: str = Field(default="VND", pattern=r"^(VND|USD|EUR)$")
    category: str
    receipt_url: Optional[str] = None

    @model_validator(mode='after')
    def high_amount_requires_receipt(self):
        """Chi phi cao phai co receipt."""
        if self.currency == "VND" and self.amount > 1000000:
            if not self.receipt_url:
                raise ValueError(
                    f"Chi phi {self.amount:,.0f} VND vuot 1M can dinh kem receipt URL"
                )
        elif self.currency == "USD" and self.amount > 50:
            if not self.receipt_url:
                raise ValueError(
                    f"Chi phi {self.amount} USD vuot 50 USD can dinh kem receipt URL"
                )
        return self

    @validator('category')
    def validate_category(cls, v):
        valid_categories = ['travel', 'meals', 'software', 'office', 'training', 'other']
        if v.lower() not in valid_categories:
            raise ValueError(f"Category phai la mot trong: {valid_categories}")
        return v.lower()

Tổng kết: Pydantic integration checklist

Pydantic + Claude Tool Use là combination hoàn hảo cho production: Claude linh hoạt trong natural language understanding, Pydantic strict về data integrity. Kết hợp hai điểm mạnh này cho ứng dụng vừa thông minh vừa reliable.

Bước tiếp theo: Tìm hiểu Tool Search với Embeddings — khi bạn có hàng nghìn tools và cần tìm đúng tool cho từng query mà không tốn hết context window.