· ai claude agent sdk · 15 min read

Dùng Claude Agent SDK để tự động tóm tắt tin tức mỗi ngày

Hướng dẫn từng bước xây dựng một agent Python đọc RSS, chọn lọc và tóm tắt tin tức bằng Claude Agent SDK - từ script đầu tiên đến custom tool, system prompt và subagent theo chủ đề.

Hướng dẫn từng bước xây dựng một agent Python đọc RSS, chọn lọc và tóm tắt tin tức bằng Claude Agent SDK - từ script đầu tiên đến custom tool, system prompt và subagent theo chủ đề.

Vấn đề

Mỗi sáng mình mở khoảng năm sáu tab: VnExpress, Tuổi Trẻ, Hacker News, vài newsletter. Đọc hết thì mất 30-40 phút, mà phần lớn nội dung không liên quan đến công việc. Bỏ qua thì lại sợ sót tin quan trọng.

Đây đúng là loại việc nên giao cho máy: đọc nhiều, lọc theo tiêu chí, viết lại ngắn gọn. Trước đây muốn làm thì phải tự viết crawler, tự parse, tự gọi API model, tự xử lý vòng lặp gọi tool. Với Claude Agent SDK, phần lớn khối lượng đó đã có sẵn.

Bài này đi từ script 10 dòng đầu tiên đến một agent hoàn chỉnh: đọc đúng nguồn bạn chỉ định, lọc theo tiêu chí, rồi ghi bản tin ra file Markdown. Toàn bộ code viết bằng Python.

Claude Agent SDK là gì?

claude-agent-sdk chính là bộ máy đằng sau Claude Code, đóng gói lại thành một thư viện Python. Bạn đưa cho nó một prompt, nó tự chạy vòng lặp agent: quyết định cần gọi tool nào, thực thi tool, đọc kết quả, gọi tiếp nếu thấy còn thiếu, rồi trả về kết luận cuối cùng.

Với bài toán điểm tin, đây đúng là thứ cần. Agent phải chủ động: gọi tool lấy tin, thấy chưa đủ thì lấy thêm nguồn khác, lọc theo tiêu chí, rồi tự ghi file kết quả. Bạn chỉ mô tả mục tiêu và cấp quyền, phần điều phối từng bước để SDK lo.

Một số tool có sẵn mà agent dùng được ngay, không cần bạn cài đặt gì:

ToolChức năng
ReadĐọc file trong thư mục làm việc
WriteTạo file mới
EditSửa chính xác một đoạn trong file có sẵn
BashChạy lệnh terminal
GlobTìm file theo pattern
GrepTìm nội dung bằng regex
WebSearchTìm kiếm web
WebFetchTải và đọc nội dung một trang web

Bước 1: Cài đặt và xác thực

Bài này dùng uv để quản lý project và phụ thuộc. Nếu máy chưa có, cài bằng một lệnh:

curl -LsSf https://astral.sh/uv/install.sh | sh

Trên Windows PowerShell:

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

Khởi tạo project và thêm phụ thuộc:

uv init news-agent
cd news-agent
uv add claude-agent-sdk httpx

SDK yêu cầu Python 3.10 trở lên, nhưng với uv thì bạn không phải tự lo chuyện đó: nó đọc requires-python trong pyproject.toml, tự tải đúng bản Python nếu máy chưa có, và tự tạo virtualenv ở .venv - không cần python -m venv rồi activate thủ công. Muốn ghim một phiên bản cụ thể thì chạy uv python pin 3.12.

httpx dùng cho phần custom tool ở Bước 3. Điểm tiện lợi khác: SDK đã đóng gói sẵn binary Claude Code cho nền tảng của bạn, nên không cần cài Claude Code riêng.

Cuối cùng là xác thực. Có hai đường đi: mở API key trả tiền theo lượng dùng, hoặc tận dụng luôn subscription Claude Code bạn đang có.

Cách 1: API key

Lấy key ở Anthropic Console rồi export ra biến môi trường:

export ANTHROPIC_API_KEY=sk-ant-xxxxx

Trên Windows PowerShell:

$env:ANTHROPIC_API_KEY = "sk-ant-xxxxx"

Cách 2: dùng subscription Claude Code sẵn có

Nếu bạn đã trả tiền cho gói Claude Pro, Max, Team hoặc Enterprise thì không cần mở thêm API key. Agent có thể xác thực bằng chính subscription đó và tiêu vào quota hiện tại của bạn, thay vì phát sinh một hoá đơn API riêng.

Cách làm là tạo một OAuth token dài hạn bằng lệnh của Claude Code:

claude setup-token

Lệnh này mở trình duyệt để bạn xác nhận, rồi in token ra terminal. Nó không tự lưu token ở đâu cả - bạn phải tự copy và đặt vào biến môi trường:

export CLAUDE_CODE_OAUTH_TOKEN=your-token

Token có hạn một năm và chỉ dùng được để gọi model. Nếu máy chưa có lệnh claude, bạn cần cài Claude Code CLI để chạy lệnh này - chỉ một lần, để lấy token.

Ba điều nên biết trước khi chọn hướng này:

ANTHROPIC_API_KEY được ưu tiên cao hơn CLAUDE_CODE_OAUTH_TOKEN. Trong thứ tự xét credential, API key đứng trên OAuth token. Nếu bạn còn sót một ANTHROPIC_API_KEY cũ trong .bashrc hay .env, agent sẽ âm thầm chạy bằng key đó và tính tiền vào đấy chứ không đụng tới subscription. Chạy unset ANTHROPIC_API_KEY trước khi dùng cách này.

Quota subscription không phải quota API. Bạn dùng chung hạn mức với Claude Code hằng ngày, nên một job điểm tin chạy nặng có thể ăn vào phần bạn để dành cho việc code. Đây là đánh đổi cần cân nhắc nếu bạn chạy agent thường xuyên.

Chỉ dùng cho agent của chính bạn. Tài liệu Agent SDK ghi rõ: trừ khi được Anthropic duyệt trước, nhà phát triển bên thứ ba không được phép cung cấp đăng nhập claude.ai hoặc rate limit của claude.ai cho sản phẩm của mình, kể cả sản phẩm xây trên Agent SDK. Nói cách khác, dùng subscription cho bản tin cá nhân như bài này thì hoàn toàn ổn; nhưng nếu định đóng gói thành sản phẩm cho người khác dùng thì phải quay lại API key. Chi tiết đầy đủ về các phương thức xác thực nằm trong tài liệu Authentication của Claude Code.

Ngoài hai cách trên, SDK còn hỗ trợ xác thực qua Amazon Bedrock, Google Cloud và Microsoft Foundry bằng các biến môi trường tương ứng.

Bước 2: Agent đầu tiên

Tạo file agent.py:

import asyncio

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query

async def main():
    async for message in query(
        prompt="Tìm 5 tin công nghệ nổi bật 24 giờ qua, tóm tắt mỗi tin 2 câu.",
        options=ClaudeAgentOptions(allowed_tools=["WebSearch", "WebFetch"]),
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)

asyncio.run(main())

Chạy:

uv run agent.py

Chỉ vậy thôi. Agent sẽ tự tìm kiếm, tự mở các trang cần đọc, rồi trả về bản tóm tắt.

Có hai chi tiết đáng chú ý trong đoạn code trên:

query() là một async generator. Nó phát ra nhiều loại message trong suốt quá trình chạy: SystemMessage (khởi tạo), AssistantMessage (mỗi lượt model phản hồi, bao gồm cả các lần gọi tool), và cuối cùng là ResultMessage. Ở đây ta chỉ quan tâm message cuối nên lọc bằng isinstance.

allowed_tools là danh sách tool được duyệt trước. Tool nằm trong danh sách sẽ chạy thẳng, không hỏi lại. Tool không có trong danh sách vẫn khả dụng nhưng phải đi qua luồng xin phép, nên hãy liệt kê đủ những tool bạn muốn agent tự dùng.

Muốn xem agent đang làm gì, thêm nhánh xử lý AssistantMessage. Đây là agent.py đầy đủ sau khi thêm:

import asyncio

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    ResultMessage,
    ToolUseBlock,
    query,
)

async def main():
    async for message in query(
        prompt="Tìm 5 tin công nghệ nổi bật 24 giờ qua, tóm tắt mỗi tin 2 câu.",
        options=ClaudeAgentOptions(allowed_tools=["WebSearch", "WebFetch"]),
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    print(f"[gọi tool] {block.name} {block.input}")
        elif isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)

asyncio.run(main())

Bước 3: Kiểm soát nguồn tin bằng custom tool

WebSearch tiện nhưng có một nhược điểm lớn với bài toán này: bạn không kiểm soát được nguồn. Hôm nay nó lấy từ VnExpress, mai có thể lấy từ một trang tổng hợp lá cải nào đó. Với bản tin đọc mỗi sáng, ta muốn nguồn cố định và biết trước.

Giải pháp là viết một custom tool đọc RSS từ danh sách nguồn do ta chỉ định. Agent SDK cho phép định nghĩa tool ngay trong process thông qua một in-process MCP server - không cần dựng server riêng.

Một tool gồm 4 phần: tên, mô tả, schema đầu vào, và handler. Tạo file news_tool.py:

import json
import re
import xml.etree.ElementTree as ET
from typing import Any

import httpx
from claude_agent_sdk import ToolAnnotations, create_sdk_mcp_server, tool

# Danh sách nguồn do bạn kiểm soát hoàn toàn
FEEDS = {
    "vnexpress": "https://vnexpress.net/rss/tin-moi-nhat.rss",
    "tuoitre": "https://tuoitre.vn/rss/tin-moi-nhat.rss",
    "hackernews": "https://hnrss.org/frontpage",
}
HEADERS = {"user-agent": "Mozilla/5.0 (news-agent)"}
TAG_RE = re.compile(r"<[^>]+>")

def clean(text: str | None) -> str:
    """Bỏ HTML lẫn trong nội dung RSS và gộp khoảng trắng thừa."""
    if not text:
        return ""
    return re.sub(r"\s+", " ", TAG_RE.sub(" ", text)).strip()

def result(text: str, is_error: bool = False) -> dict[str, Any]:
    """Gói một chuỗi thành tool result đúng định dạng SDK."""
    out: dict[str, Any] = {"content": [{"type": "text", "text": text}]}
    if is_error:
        out["is_error"] = True
    return out

@tool(
    "get_headlines",
    "Lấy tin mới nhất từ một nguồn RSS đã cấu hình sẵn. limit mặc định 15.",
    {
        "type": "object",
        "properties": {
            "source": {"type": "string", "enum": list(FEEDS)},
            "limit": {"type": "integer", "minimum": 1, "maximum": 30},
        },
        "required": ["source"],
    },
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def get_headlines(args: dict[str, Any]) -> dict[str, Any]:
    source, limit = args["source"], args.get("limit", 15)
    try:
        async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client:
            response = await client.get(FEEDS[source], headers=HEADERS)
        if response.status_code != 200:
            return result(f"{source} trả về HTTP {response.status_code}", is_error=True)

        root = ET.fromstring(response.content)
        items = [
            {
                "title": clean(item.findtext("title")),
                "link": clean(item.findtext("link")),
                "published_at": clean(item.findtext("pubDate")),
                "summary": clean(item.findtext("description"))[:400],
            }
            for item in list(root.iter("item"))[:limit]
        ]
        return result(json.dumps(items, ensure_ascii=False, indent=2))
    except Exception as e:
        # Tự soạn thông báo lỗi để Claude biết đường xử lý tiếp
        return result(f"Không lấy được tin từ {source}: {e}", is_error=True)

news_server = create_sdk_mcp_server(name="news", version="1.0.0", tools=[get_headlines])

Bốn điểm đáng chú ý:

Tên tool có namespace. Server đăng ký với key news, nên get_headlines có tên đầy đủ là mcp__news__get_headlines (mẫu mcp__{server}__{tool}). Đây là tên bạn điền vào allowed_tools; nhiều tool thì dùng mcp__news__*.

Dùng JSON Schema đầy đủ vì cần enum và tham số tuỳ chọn. Dạng rút gọn {"source": str, "limit": int} tiện hơn, nhưng nó coi mọi key là bắt buộc và không diễn đạt được enum. Ở đây ta cần cả hai, nên phải khai báo required tường minh rồi đọc giá trị tuỳ chọn bằng args.get("limit", 15). Danh sách enum lấy thẳng từ list(FEEDS) để thêm nguồn mới chỉ phải sửa một chỗ.

Trả is_error: True thay vì để exception văng ra. Exception không làm agent dừng - SDK vẫn bắt và chuyển thành kết quả lỗi, nhưng Claude chỉ nhận được message thô. Tự bắt lỗi giúp bạn nói rõ nguồn nào hỏng, để Claude bỏ qua nguồn đó và đi tiếp với nguồn còn lại. Mọi kết quả trả về đều đi qua hàm result() cho khỏi lặp lại cấu trúc content ở từng nhánh.

readOnlyHint=True cho phép gọi song song. Claude biết tool không thay đổi gì nên lấy cả ba nguồn cùng lúc thay vì tuần tự. Lưu ý ToolAnnotations dùng camelCase, khác với phần còn lại của API Python.

Về parser: xml.etree.ElementTree trong thư viện chuẩn tự xử lý CDATA, nên phần description của VnExpress đọc ra bình thường, chỉ còn phải gỡ mấy thẻ HTML bên trong. Code này hợp với RSS 2.0; nếu cần đọc cả Atom (thẻ entry thay vì item) thì dùng feedparser.

Giờ nối tool vào agent:

import asyncio

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query
from news_tool import news_server

async def main():
    options = ClaudeAgentOptions(
        mcp_servers={"news": news_server},
        allowed_tools=["mcp__news__get_headlines"],
    )
    async for message in query(
        prompt="Lấy tin từ cả ba nguồn và tóm tắt 5 tin đáng chú ý nhất.",
        options=options,
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)

asyncio.run(main())

Bước 4: Định dạng đầu ra và ghi ra file

Đến đây agent đã lấy đúng nguồn, nhưng định dạng đầu ra mỗi ngày một khác. Hai thứ cần thêm: một system prompt mô tả rõ tiêu chí và bố cục, và tool Write để ghi kết quả ra file.

System prompt sẽ được dùng lại ở Bước 5, nên tách hẳn ra file prompts.py:

SYSTEM_PROMPT = """Bạn là trợ lý điểm tin cho một lập trình viên Việt Nam.

Tiêu chí chọn tin:
- Ưu tiên: công nghệ, kinh tế vĩ mô, chính sách ảnh hưởng đến ngành phần mềm.
- Bỏ qua: showbiz, thể thao, tin giật gân, tin quảng cáo trá hình.
- Nếu nhiều nguồn đưa cùng một sự kiện, gộp lại thành một mục duy nhất.

Định dạng đầu ra (Markdown):
- Tiêu đề cấp 1: "Điểm tin ngày <ngày>"
- Mỗi tin là một mục cấp 2, gồm: tiêu đề tin, 2-3 câu tóm tắt, và link nguồn.
- Cuối bài thêm mục "Đáng chú ý nhất" giải thích trong 1 câu vì sao tin đó quan trọng.

Chỉ dùng thông tin có trong kết quả tool. Không suy đoán, không bịa chi tiết
không xuất hiện trong nguồn. Nếu một nguồn lỗi, ghi rõ ở cuối bài."""

Rồi agent.py:

import asyncio
from datetime import date

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query
from news_tool import news_server
from prompts import SYSTEM_PROMPT

async def main():
    today = date.today().isoformat()
    options = ClaudeAgentOptions(
        system_prompt=SYSTEM_PROMPT,
        mcp_servers={"news": news_server},
        allowed_tools=["mcp__news__get_headlines", "Write"],
        max_turns=20,
        max_budget_usd=0.5,
    )
    async for message in query(
        prompt=(
            "Lấy tin mới nhất từ cả ba nguồn, chọn 5-7 tin theo tiêu chí, "
            f"rồi ghi bản tin vào file ./ban-tin/{today}.md"
        ),
        options=options,
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)

asyncio.run(main())

Hai option mới đáng chú ý:

max_turns giới hạn số lượt agent được chạy. Nếu vì lý do nào đó agent rơi vào vòng lặp gọi tool liên tục, nó sẽ dừng sau 20 lượt thay vì chạy mãi.

max_budget_usd đặt trần chi phí cho một lần chạy - lưới an toàn cho ví tiền khi có sự cố ngoài dự tính.

Câu cuối trong system prompt - yêu cầu chỉ dùng thông tin có trong kết quả tool - không phải thừa. Model hoàn toàn có kiến thức nền về các chủ đề tin tức, và nếu không ràng buộc, nó có thể bổ sung chi tiết “nghe hợp lý” nhưng không có trong nguồn bạn cung cấp. Với bản tin thì đó là lỗi nghiêm trọng.

Bước 5: Dùng subagent để tách chủ đề

Khi số nguồn tăng lên, nhồi hết vào một agent duy nhất sẽ khiến context phình to và chất lượng tóm tắt giảm. Cách xử lý là giao mỗi mảng chủ đề cho một subagent riêng, mỗi subagent có context độc lập, rồi agent chính tổng hợp lại.

import asyncio
from datetime import date

from claude_agent_sdk import AgentDefinition, ClaudeAgentOptions, ResultMessage, query
from news_tool import news_server
from prompts import SYSTEM_PROMPT

COLLECTORS = {
    "tin-trong-nuoc": AgentDefinition(
        description="Chuyên trách tin tức trong nước từ báo Việt Nam.",
        prompt=(
            "Lấy tin từ nguồn vnexpress và tuoitre. Loại bỏ showbiz, thể thao. "
            "Trả về tối đa 4 tin, mỗi tin gồm tiêu đề, 2 câu tóm tắt và link."
        ),
        tools=["mcp__news__get_headlines"],
    ),
    "tin-cong-nghe": AgentDefinition(
        description="Chuyên trách tin công nghệ quốc tế.",
        prompt=(
            "Lấy tin từ nguồn hackernews. Ưu tiên tin về AI, ngôn ngữ lập trình "
            "và hạ tầng. Trả về tối đa 4 tin kèm link, tóm tắt bằng tiếng Việt."
        ),
        tools=["mcp__news__get_headlines"],
    ),
}

async def main():
    today = date.today().isoformat()
    options = ClaudeAgentOptions(
        system_prompt=SYSTEM_PROMPT,
        mcp_servers={"news": news_server},
        allowed_tools=["mcp__news__get_headlines", "Write", "Agent"],
        agents=COLLECTORS,
        max_turns=30,
        max_budget_usd=1.0,
    )
    try:
        async for message in query(
            prompt=(
                "Dùng subagent tin-trong-nuoc và tin-cong-nghe để thu thập, sau đó "
                f"tổng hợp thành một bản tin duy nhất và ghi vào ./ban-tin/{today}.md"
            ),
            options=options,
        ):
            if isinstance(message, ResultMessage) and message.subtype == "success":
                print(message.result)
    except Exception as error:
        # query() raise sau khi phát ra result lỗi, vòng lặp chỉ in bản thành công
        print(f"Chạy thất bại: {error}")

asyncio.run(main())

Điểm dễ sai nhất ở đây: subagent được gọi thông qua tool Agent, nên bạn phải thêm "Agent" vào allowed_tools. Thiếu dòng này thì agent chính sẽ bị chặn lại ở bước xin phép và job tự động sẽ đứng im.

Agent chính vẫn giữ nguyên SYSTEM_PROMPT từ Bước 4 - nó là bên chịu trách nhiệm tổng hợp và ghi file, nên vẫn cần đủ tiêu chí chọn tin và quy tắc định dạng. Hai subagent thì có prompt riêng, hẹp hơn, chỉ mô tả việc thu thập. Đây là cách phân vai: subagent gom nguyên liệu, agent chính quyết định bản tin trông ra sao.

Mỗi subagent có danh sách tools riêng. Ở trên, cả hai subagent chỉ được cấp quyền đọc RSS - chúng không ghi được file. Chỉ agent chính mới có Write. Đây là cách phân quyền tốt: subagent thu thập, agent chính chịu trách nhiệm ghi kết quả cuối cùng.

Một chi tiết dễ vấp: AgentDefinition đặt tên các field tuỳ chọn theo camelCase (disallowedTools, maxTurns, permissionMode, mcpServers), trong khi ClaudeAgentOptions lại dùng snake_case. Viết max_turns=5 bên trong AgentDefinition sẽ báo lỗi, phải là maxTurns=5.

Ngoài ra, các message phát ra từ trong một subagent sẽ có trường parent_tool_use_id, giúp bạn biết message nào thuộc lần chạy subagent nào khi cần log chi tiết.

Kết luận

Điều đáng giá nhất của Claude Agent SDK trong bài toán này không nằm ở khả năng tóm tắt, mà ở chỗ bạn không phải viết vòng lặp điều phối tool, không phải tự xử lý việc agent cần gọi thêm dữ liệu giữa chừng, và có sẵn các tool đọc/ghi file để nối kết quả vào phần còn lại của hệ thống.

Tóm lại lộ trình đã đi qua:

  1. Agent tối giản với WebSearch - chạy được ngay nhưng không kiểm soát nguồn.
  2. Custom tool đọc RSS qua in-process MCP server - nguồn tin do bạn quyết định.
  3. System prompt và tool Write - đầu ra ổn định, ghi thẳng ra Markdown.
  4. Subagent theo chủ đề - context tách bạch, phân quyền rõ ràng.

Từ khung này bạn có thể mở rộng theo nhiều hướng: thay Write bằng một custom tool đẩy bản tin lên Slack hay Telegram, thêm nguồn newsletter qua IMAP, hoặc dùng hook PostToolUse để ghi log mọi lần gọi tool phục vụ việc kiểm toán.

Tài liệu chính thức để tham khảo thêm: Agent SDK overview, tham chiếu Python và hướng dẫn custom tools.

Share:
Back to Blog