Bốn bài trước trong module này đã trang bị cho bạn bốn "vũ khí" riêng lẻ: nén tool schema (giảm token cho phần định nghĩa tool — tool definition), dynamic tool loading (chỉ nạp tool cần thiết cho từng task), trimming kết quả trả về, và batching/chaining tool call. Vấn đề là trong thực tế, các kỹ thuật này hiếm khi được áp dụng đơn lẻ — một MCP server (Model Context Protocol server — máy chủ triển khai chuẩn giao tiếp MCP giữa agent và tool) sản xuất thực sự cần cả bốn cùng lúc để đạt hiệu quả tối đa.
Bài lab này sẽ đưa bạn đi qua một quy trình tối ưu hoàn chỉnh, từ đầu đến cuối, trên một MCP server giả lập cho một "DevOps Toolkit" — bộ công cụ mô phỏng các tác vụ quản lý repository, pull request, issue, CI/CD mà một agent hỗ trợ kỹ sư phần mềm thường cần dùng. Bạn sẽ tự tay dựng server bằng Node.js với @modelcontextprotocol/sdk, viết script Python để đo token bằng tiktoken, đo baseline, áp từng kỹ thuật tối ưu một, và đo lại sau mỗi bước để có số liệu định lượng rõ ràng. Đây là bài lab tổng hợp và dài nhất của module — hãy chuẩn bị khoảng 60-90 phút để làm theo đầy đủ.
Lưu ý quan trọng trước khi bắt đầu: tiktoken (bộ tokenizer của OpenAI, dùng encoding cl100k_base) không phải là tokenizer chính xác của Claude hay các model khác. Chúng ta dùng nó vì nó miễn phí, chạy offline, và cho ra con số tỷ lệ tương đối đủ tin cậy để so sánh trước/sau. Muốn có số liệu tuyệt đối chính xác với Claude, bạn cần dùng endpoint count_tokens của Anthropic API — bài lab sẽ nhắc lại điều này ở phần mở rộng.
Chuẩn Bị Môi Trường Và Đo Baseline
Trước khi tối ưu bất cứ thứ gì, chúng ta cần một con số baseline (mức nền) đáng tin cậy. Không có baseline, mọi phần trăm cải thiện sau này chỉ là cảm tính.
Tạo thư mục lab và project Node.js
Mở terminal, tạo thư mục lab và khởi tạo project Node:
mkdir -p ~/labs/mcp-token-optimization
cd ~/labs/mcp-token-optimization
npm init -y
npm install @modelcontextprotocol/sdk zod
Sau khi cài xong, mở package.json và thêm "type": "module" để dùng cú pháp ES module (import/export) xuyên suốt bài lab:
{
"name": "mcp-token-optimization",
"version": "1.0.0",
"type": "module",
"dependencies": {
"@modelcontextprotocol/sdk": "^1.0.0",
"zod": "^3.23.0"
}
}
Định nghĩa tool set baseline (verbose — dài dòng, chưa tối ưu)
Tạo file tool-definitions.js. Đây là nơi khai báo toàn bộ tool theo đúng format mà MCP server trả về trong response của tools/list (tương đương ListTools) — chính là phần payload mà mọi agent phải "nuốt" vào context ở mỗi lần gọi model có kèm tool.
Để mô phỏng một tình huống thực tế — nơi tool definition được viết bởi nhiều người, qua nhiều lần chỉnh sửa, và không ai để tâm đến chi phí token — chúng ta viết description dài dòng, lặp lại thông tin đã có trong schema, và thêm ví dụ ngay trong description:
// tool-definitions.js
// BASELINE — verbose, unoptimized tool definitions.
export const TOOLS = [
{
name: "list_pull_requests",
description:
"This tool allows you to list all pull requests that currently exist " +
"in a given repository on the DevOps platform. Use this tool whenever " +
"the user asks about pull requests, PRs, code review status, or wants " +
"to know what changes are currently open for review in a repository. " +
"The tool will return a list of pull request objects, each containing " +
"detailed information such as the pull request ID, title, description, " +
"current state (open, closed, or merged), the author who created it, " +
"the branch it originates from, the target branch, creation date, last " +
"updated date, number of comments, number of approvals, and a URL " +
"pointing to the pull request on the web platform. For example, if the " +
"user asks 'what PRs are open in the billing-service repo', you should " +
"call this tool with repo='billing-service' and state='open'.",
inputSchema: {
type: "object",
properties: {
repo: {
type: "string",
description:
"The name of the repository to list pull requests from. " +
"This should be the short repository name, not a full URL. " +
"Example values: 'billing-service', 'auth-gateway'.",
},
state: {
type: "string",
enum: ["open", "closed", "merged", "all"],
description:
"Filter pull requests by their current state. Use 'open' for " +
"pull requests that are still awaiting review or merge. Use " +
"'closed' for pull requests that were closed without merging. " +
"Use 'merged' for pull requests that have already been merged " +
"into the target branch. Use 'all' to return every pull request " +
"regardless of state. Defaults to 'open' if not specified.",
},
limit: {
type: "number",
description:
"Maximum number of pull requests to return in the response. " +
"Useful for pagination when a repository has many pull requests. " +
"Defaults to 25 if not specified. Maximum allowed value is 100.",
},
},
required: ["repo"],
},
},
{
name: "get_pull_request",
description:
"This tool retrieves the full, detailed information for a single, " +
"specific pull request identified by its repository and its pull " +
"request number. Use this when the user wants to inspect one " +
"particular pull request in detail, for example to read its full " +
"description, see the list of files changed, or check its review " +
"status. This is different from list_pull_requests, which returns " +
"a summary list of many pull requests — this tool returns one full " +
"object with every available field, including the complete diff " +
"statistics and the list of reviewers.",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name, e.g. 'billing-service'." },
pr_number: { type: "number", description: "The numeric identifier of the pull request within the repository, as shown in the platform URL." },
},
required: ["repo", "pr_number"],
},
},
{
name: "list_issues",
description:
"Lists issues (bug reports, feature requests, or tasks tracked in " +
"the issue tracker) that exist for a given repository. Use this " +
"whenever the user asks about bugs, tickets, tasks, or backlog items. " +
"Returns a list of issue objects including ID, title, labels, " +
"assignee, priority, and current status (open or closed).",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name." },
status: {
type: "string",
enum: ["open", "closed", "all"],
description: "Filter issues by status. Defaults to 'open'.",
},
label: { type: "string", description: "Optional label to filter issues by, e.g. 'bug', 'p0', 'needs-triage'." },
},
required: ["repo"],
},
},
{
name: "create_issue",
description:
"Creates a brand new issue in the specified repository's issue " +
"tracker. Use this tool when the user explicitly asks to file a bug, " +
"open a ticket, or create a task. Always confirm the title and " +
"description with the user before calling this tool, since issue " +
"creation is a write operation that cannot be easily undone through " +
"this toolset.",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name to create the issue in." },
title: { type: "string", description: "A short, descriptive title for the issue." },
body: { type: "string", description: "The full description/body text of the issue, in Markdown." },
labels: { type: "array", items: { type: "string" }, description: "Optional list of labels to attach to the issue on creation." },
},
required: ["repo", "title"],
},
},
{
name: "get_ci_status",
description:
"Retrieves the current continuous integration (CI) pipeline status " +
"for a given commit or branch. Use this when the user asks whether " +
"the build is passing, whether tests are green, or whether it is " +
"safe to merge. Returns the overall status plus a breakdown per " +
"pipeline stage (lint, test, build, deploy).",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name." },
ref: { type: "string", description: "Branch name or commit SHA to check CI status for." },
},
required: ["repo", "ref"],
},
},
{
name: "list_deployments",
description:
"Lists recent deployments for a given repository and environment. " +
"Use this when the user asks what version is currently deployed, " +
"or wants a history of recent deploys to production or staging.",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name." },
environment: { type: "string", enum: ["staging", "production"], description: "Which environment to list deployments for." },
},
required: ["repo", "environment"],
},
},
{
name: "get_deployment_logs",
description:
"Fetches the full raw log output produced during a specific " +
"deployment. Use this when the user needs to debug a failed " +
"deployment or wants to inspect what happened during a deploy. " +
"Logs can be very long for complex deployments involving many services.",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name." },
deployment_id: { type: "string", description: "The unique identifier of the deployment to fetch logs for." },
},
required: ["repo", "deployment_id"],
},
},
{
name: "search_code",
description:
"Performs a full-text search across the source code of a repository. " +
"Use this whenever the user wants to find where a function, variable, " +
"string, or pattern is used or defined across the codebase.",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name." },
query: { type: "string", description: "The text or pattern to search for." },
},
required: ["repo", "query"],
},
},
{
name: "list_commits",
description:
"Lists recent commits on a given branch of a repository, including " +
"commit SHA, author, message, and timestamp. Use this to inspect " +
"recent history or find when a particular change was introduced.",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name." },
branch: { type: "string", description: "Branch name to list commits from. Defaults to the default branch." },
},
required: ["repo"],
},
},
{
name: "list_repositories",
description:
"Lists all repositories that the current user or organization has " +
"access to. Use this when the user asks 'what repos do we have' or " +
"needs to discover the name of a repository before calling other tools.",
inputSchema: { type: "object", properties: {}, required: [] },
},
{
name: "get_repository",
description:
"Retrieves detailed metadata about a single repository, including " +
"its description, default branch, primary language, and visibility " +
"(public or private).",
inputSchema: {
type: "object",
properties: { repo: { type: "string", description: "Repository short name." } },
required: ["repo"],
},
},
{
name: "get_issue",
description:
"Retrieves full detail for a single specific issue by its number, " +
"including its complete description, all comments, and history of " +
"status changes.",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name." },
issue_number: { type: "number", description: "The numeric identifier of the issue." },
},
required: ["repo", "issue_number"],
},
},
];
// HANDLERS — mock implementations returning realistic-shaped fake data.
export const HANDLERS = {
list_pull_requests: async ({ repo, state = "open", limit = 25 }) => {
return Array.from({ length: Math.min(limit, 6) }, (_, i) => ({
id: 1000 + i,
title: `Fix edge case in ${repo} handler #${i}`,
state,
user: { login: "dev-user", id: 42, avatar_url: "https://example.com/avatar.png" },
created_at: "2026-07-01T10:00:00Z",
updated_at: "2026-07-05T14:22:00Z",
html_url: `https://devops.example.com/${repo}/pull/${1000 + i}`,
comments: 3,
review_comments: 5,
commits: 4,
additions: 120,
deletions: 34,
changed_files: 6,
base: { ref: "main" },
head: { ref: `feature/fix-${i}` },
}));
},
get_deployment_logs: async ({ deployment_id }) => {
const lines = Array.from({ length: 400 }, (_, i) => `[2026-07-09T10:${String(i).padStart(2, "0")}:00Z] step-${i} INFO ok`);
return { deployment_id, log_text: lines.join("\n") };
},
// ... other handlers follow the same pattern (omitted here for brevity,
// see full source in the lab repository).
};
Sao chép dài dòng như trên là có chủ đích — nó mô phỏng chính xác kiểu tool definition bạn sẽ gặp trong các MCP server viết vội, nơi mỗi người thêm một câu "for example..." mà không ai review lại tổng thể.
Viết server MCP thực thi bằng stdio transport
Tạo file server.js, dùng lớp Server cấp thấp của SDK để trả TOOLS nguyên bản qua tools/list và định tuyến tools/call tới HANDLERS:
// server.js
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { ListToolsRequestSchema, CallToolRequestSchema } from "@modelcontextprotocol/sdk/types.js";
import { TOOLS, HANDLERS } from "./tool-definitions.js";
const server = new Server(
{ name: "devops-mcp-server", version: "1.0.0" },
{ capabilities: { tools: {} } }
);
server.setRequestHandler(ListToolsRequestSchema, async () => ({ tools: TOOLS }));
server.setRequestHandler(CallToolRequestSchema, async (request) => {
const handler = HANDLERS[request.params.name];
if (!handler) throw new Error(`Unknown tool: ${request.params.name}`);
const result = await handler(request.params.arguments ?? {});
return { content: [{ type: "text", text: JSON.stringify(result) }] };
});
const transport = new StdioServerTransport();
await server.connect(transport);
Chạy thử nhanh để chắc chắn server không lỗi cú pháp:
node server.js
Trích xuất tool definitions ra JSON để đo
Để đo token mà không cần dựng một MCP client đầy đủ, ta viết một script nhỏ "dump" thẳng nội dung của TOOLS — chính là những gì client sẽ nhận được khi gọi tools/list:
// dump-tools.js
import { writeFileSync } from "fs";
import { TOOLS } from "./tool-definitions.js";
const outputPath = process.argv[2] || "tools_baseline.json";
writeFileSync(outputPath, JSON.stringify(TOOLS, null, 2));
console.log(`Wrote ${TOOLS.length} tool definitions to ${outputPath}`);
Chạy:
node dump-tools.js tools_baseline.json
Thiết lập Python và script đo token
Tạo virtual environment và cài tiktoken:
python3 -m venv .venv
source .venv/bin/activate
pip install tiktoken
Viết measure_baseline.py. Mục tiêu của script này là tách riêng phần token do tool definitions gây ra khỏi phần token của nội dung hội thoại thật — bằng cách ghép tool definitions với một user message tối thiểu (chỉ "Hi"), để mọi token đo được ngoài vài token của "Hi" đều là overhead do tool gây ra:
import json
import sys
import tiktoken
encoding = tiktoken.get_encoding("cl100k_base")
def count_tokens(text: str) -> int:
return len(encoding.encode(text))
def main(tools_path: str) -> None:
with open(tools_path, "r", encoding="utf-8") as f:
tools = json.load(f)
# This mirrors the "tools" field of the actual API request payload.
tools_json = json.dumps(tools, ensure_ascii=False)
tool_def_tokens = count_tokens(tools_json)
minimal_user_message = "Hi"
user_tokens = count_tokens(minimal_user_message)
print(f"Tool set : {tools_path}")
print(f"Number of tools : {len(tools)}")
print(f"Tool definitions (tokens) : {tool_def_tokens:,}")
print(f"Minimal user msg (tokens) : {user_tokens}")
print(f"Total request overhead : {tool_def_tokens + user_tokens:,}")
print(f"Avg tokens / tool : {tool_def_tokens / len(tools):.1f}")
if __name__ == "__main__":
main(sys.argv[1] if len(sys.argv) > 1 else "tools_baseline.json")
Chạy:
python measure_baseline.py tools_baseline.json
Kết quả mẫu (số liệu minh họa, dùng để so sánh xuyên suốt bài lab):
Tool set : tools_baseline.json
Number of tools : 12
Tool definitions (tokens) : 3,820
Minimal user msg (tokens) : 1
Total request overhead : 3,821
Avg tokens / tool : 318.3
3.820 token chỉ để "giới thiệu" 12 tool cho model — trước khi model đọc một dòng yêu cầu nào từ người dùng. Nếu agent của bạn gọi model 10 lần trong một session mà không có prompt caching hiệu quả, riêng phần này đã tốn khoảng 38.000 token, chưa tính lịch sử hội thoại hay kết quả tool.
Mẹo
Luôn đo token của tool definitions tách biệt khỏi token của hội thoại thực. Nhiều team chỉ nhìn vào tổng token của cả session và bỏ lỡ việc overhead tool chiếm bao nhiêu phần trăm — cách ghép với một user message tối thiểu như trên là thủ thuật đơn giản nhất để cô lập chỉ số này mà không cần chỉnh sửa gì ở phía model.
Bước 1: Nén Tool Schema
Nhìn lại tools_baseline.json, có ba "thủ phạm" chính gây phình token: (1) description lặp lại thông tin đã nằm trong schema (ví dụ nói rõ enum values bằng văn xuôi trong khi schema đã có enum), (2) câu mở đầu sáo rỗng kiểu "This tool allows you to...", và (3) ví dụ cụ thể nhúng thẳng vào description thay vì để model tự suy luận từ tên tham số và enum.
Sửa lại tool-definitions.js, giữ nguyên tên tool, tên tham số, và ngữ nghĩa — chỉ nén cách diễn đạt:
// tool-definitions.js — Step 1: compressed descriptions and schema.
export const TOOLS = [
{
name: "list_pull_requests",
description: "List pull requests in a repository, optionally filtered by state.",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name." },
state: { type: "string", enum: ["open", "closed", "merged", "all"], description: "Default: open." },
limit: { type: "number", description: "Max results, default 25, max 100." },
},
required: ["repo"],
},
},
{
name: "get_pull_request",
description: "Get full detail of one pull request by repo and number.",
inputSchema: {
type: "object",
properties: {
repo: { type: "string", description: "Repository short name." },
pr_number: { type: "number" },
},
required: ["repo", "pr_number"],
},
},
{
name: "list_issues",
description: "List issues in a repository, optionally filtered by status/label.",
inputSchema: {
type: "object",
properties: {
repo: { type: "string" },
status: { type: "string", enum: ["open", "closed", "all"], description: "Default: open." },
label: { type: "string" },
},
required: ["repo"],
},
},
{
name: "create_issue",
description: "Create a new issue in a repository's tracker (write operation — confirm with user first).",
inputSchema: {
type: "object",
properties: {
repo: { type: "string" },
title: { type: "string" },
body: { type: "string" },
labels: { type: "array", items: { type: "string" } },
},
required: ["repo", "title"],
},
},
// ... remaining tools compressed the same way (see full file in the lab repo).
];
Nguyên tắc áp dụng nhất quán cho toàn bộ 12 tool: mỗi description không quá một câu; enum tự giải thích qua giá trị của nó (không diễn giải lại bằng văn xuôi); loại bỏ mọi câu ví dụ trong description (model đủ khả năng suy luận từ tên tham số rõ nghĩa như repo, state, limit).
Dump lại và đo:
node dump-tools.js tools_step1.json
Viết measure_step1.py — tái sử dụng gần như toàn bộ logic của measure_baseline.py, chỉ khác biệt là in thêm dòng so sánh với baseline:
import json
import tiktoken
encoding = tiktoken.get_encoding("cl100k_base")
def count_tokens(text: str) -> int:
return len(encoding.encode(text))
def load_tool_tokens(path: str) -> tuple[int, int]:
with open(path, "r", encoding="utf-8") as f:
tools = json.load(f)
tokens = count_tokens(json.dumps(tools, ensure_ascii=False))
return tokens, len(tools)
def main() -> None:
baseline_tokens, baseline_count = load_tool_tokens("tools_baseline.json")
step1_tokens, step1_count = load_tool_tokens("tools_step1.json")
reduction = (1 - step1_tokens / baseline_tokens) * 100
print(f"Baseline : {baseline_count} tools, {baseline_tokens:,} tokens")
print(f"Step 1 : {step1_count} tools, {step1_tokens:,} tokens")
print(f"Reduction: {reduction:.1f}%")
if __name__ == "__main__":
main()
python measure_step1.py
Kết quả mẫu:
Baseline : 12 tools, 3,820 tokens
Step 1 : 12 tools, 1,640 tokens
Reduction: 57.1%
Chỉ bằng việc viết lại description và loại bỏ trùng lặp — không đổi tên tool, không đổi tham số, không mất khả năng của agent — chúng ta đã cắt hơn nửa số token cho tool definitions. Đây thường là "quick win" rẻ nhất trong toàn bộ quá trình tối ưu, vì không cần đổi kiến trúc gì cả.
Mẹo
Sau khi nén, luôn chạy lại một bộ test case thực tế của agent (ví dụ 10-15 câu hỏi mẫu người dùng thường hỏi) để chắc chắn model vẫn chọn đúng tool và đúng tham số. Mất một câu ví dụ tưởng chừng vô hại trong description đôi khi khiến model chọn sai tool cho các case hiếm gặp — đo token không có ý nghĩa nếu đổi lại tỷ lệ chọn sai tool tăng lên.
Bước 2: Thêm Dynamic Tool Filtering Ở Tầng Client
Nén schema giúp mỗi tool "nhẹ" hơn, nhưng ta vẫn gửi đủ 12 tool cho mọi request, bất kể user đang hỏi về pull request hay đang hỏi về deployment. Dynamic tool filtering (lọc tool động) giải quyết đúng vấn đề này: phân loại ý định của request, rồi chỉ gửi tập con tool liên quan.
Tạo mcp_filtered_client.py. Trước tiên định nghĩa các nhóm tool theo loại tác vụ — đây là bước quan trọng nhất, cần dựa trên cách người dùng thực sự sử dụng agent của bạn (khảo sát log thực tế nếu có, đừng đoán):
import json
TOOL_GROUPS: dict[str, list[str]] = {
"code_review": ["list_pull_requests", "get_pull_request", "search_code", "list_commits"],
"issue_triage": ["list_issues", "get_issue", "create_issue"],
"deployment_ops": ["get_ci_status", "list_deployments", "get_deployment_logs"],
"repo_discovery": ["list_repositories", "get_repository"],
}
INTENT_KEYWORDS: dict[str, list[str]] = {
"code_review": ["pull request", "pr ", "diff", "commit", "code review", "review"],
"issue_triage": ["issue", "bug", "ticket", "ticket", "backlog"],
"deployment_ops": ["deploy", "ci", "pipeline", "logs", "build status"],
"repo_discovery": ["repository", "which repos", "repo list", "list repos"],
}
def classify_intent(user_message: str) -> str:
"""Very simple keyword-based classifier. In production, consider an
LLM-based or embedding-based classifier for better accuracy on
ambiguous or multi-intent messages."""
text = user_message.lower()
scores = {
group: sum(1 for kw in keywords if kw in text)
for group, keywords in INTENT_KEYWORDS.items()
}
best_group = max(scores, key=scores.get)
return best_group if scores[best_group] > 0 else "code_review" # safe fallback
def get_filtered_tools(all_tools: list[dict], group: str) -> list[dict]:
allowed_names = set(TOOL_GROUPS[group])
return [tool for tool in all_tools if tool["name"] in allowed_names]
def build_tools_payload(all_tools_path: str, user_message: str) -> list[dict]:
with open(all_tools_path, "r", encoding="utf-8") as f:
all_tools = json.load(f)
group = classify_intent(user_message)
filtered = get_filtered_tools(all_tools, group)
print(f"[dynamic-filter] intent={group} -> sending {len(filtered)}/{len(all_tools)} tools")
return filtered
if __name__ == "__main__":
sample_message = "Can you check if there are any open PRs waiting for review in billing-service?"
tools_to_send = build_tools_payload("tools_step1.json", sample_message)
print(json.dumps([t["name"] for t in tools_to_send], indent=2))
Chạy thử với vài câu mẫu khác nhau để kiểm chứng bộ phân loại:
python mcp_filtered_client.py
[dynamic-filter] intent=code_review -> sending 4/12 tools
[
"list_pull_requests",
"get_pull_request",
"search_code",
"list_commits"
]
Từ 12 tool (1.640 token sau khi nén ở bước 1) xuống 4 tool cho một request cụ thể. Ước lượng nhanh: nếu 4 tool trung bình chiếm khoảng 1/3 tổng token (tỷ lệ tương ứng số lượng), ta còn khoảng 550-650 token/request cho phần tool definitions — giảm tiếp gần 60-65% so với kết quả sau bước 1, và tổng cộng khoảng 83-85% so với baseline ban đầu.
Có hai rủi ro cần lưu ý với kỹ thuật này. Thứ nhất, bộ phân loại từ khóa đơn giản sẽ sai với các request đa ý định ("check PR status rồi tạo issue cho lỗi tìm thấy") — với hệ thống production, nên dùng phân loại bằng embedding hoặc một lệnh gọi LLM nhỏ, rẻ (model nhỏ, ít token) làm bước tiền xử lý. Thứ hai, luôn cần một nhóm "fallback" hoặc cơ chế fallback về full tool set khi độ tin cậy phân loại thấp, để không chặn agent khỏi tool nó thực sự cần.
Mẹo
Log lại mọi lần agent "muốn gọi một tool không có trong danh sách đã filter" (model sẽ báo lỗi hoặc cố gọi tên tool không tồn tại) — đây là tín hiệu trực tiếp cho thấy bộ phân loại intent của bạn đang lọc sai nhóm, và là nguồn dữ liệu tốt nhất để cải thiện INTENT_KEYWORDS hoặc chuyển sang phân loại bằng model.
Bước 3: Trim Kết Quả Trả Về Trong Tool Handler
Giảm token ở phần tool definitions chỉ là một nửa vấn đề — nửa còn lại nằm ở kết quả mà tool trả về. Nhìn lại HANDLERS.get_deployment_logs trong baseline: nó trả về toàn bộ 400 dòng log thô. Trong khi đó, hầu hết câu hỏi thực tế ("deploy này có lỗi gì") chỉ cần vài chục dòng cuối hoặc các dòng chứa từ khóa lỗi.
Tạo file trim-helpers.js chứa các hàm rút gọn kết quả theo từng loại tool:
// trim-helpers.js
export function trimPullRequest(pr) {
return {
id: pr.id,
title: pr.title,
state: pr.state,
author: pr.user?.login,
created_at: pr.created_at,
changed_files: pr.changed_files,
url: pr.html_url,
};
}
export function trimDeploymentLogs(logText, maxLines = 50) {
const lines = logText.split("\n");
if (lines.length <= maxLines) return logText;
const errorLines = lines.filter((l) => /error|fail|exception/i.test(l));
const tail = lines.slice(-maxLines);
const omittedCount = lines.length - maxLines;
const summary = [
`(${lines.length} total lines; showing last ${maxLines} plus any error lines; ${omittedCount} lines omitted)`,
];
if (errorLines.length > 0) {
summary.push("--- lines matching error/fail/exception ---", ...errorLines.slice(0, 20));
}
summary.push("--- last lines ---", ...tail);
return summary.join("\n");
}
Cập nhật HANDLERS trong tool-definitions.js để dùng các hàm trim này trước khi trả kết quả:
import { trimPullRequest, trimDeploymentLogs } from "./trim-helpers.js";
export const HANDLERS = {
list_pull_requests: async ({ repo, state = "open", limit = 25 }) => {
const rawResults = /* ... fetch or mock raw PR objects as before ... */ [];
return rawResults.map(trimPullRequest);
},
get_deployment_logs: async ({ deployment_id }) => {
const raw = /* ... fetch or mock raw 400-line log text ... */ "";
return { deployment_id, log_text: trimDeploymentLogs(raw, 50) };
},
// ...
};
Đo nhanh sự khác biệt bằng một script Python đơn giản đo độ dài chuỗi JSON trước/sau trim, dùng cùng hàm count_tokens đã có:
import json
import tiktoken
encoding = tiktoken.get_encoding("cl100k_base")
def count_tokens(text: str) -> int:
return len(encoding.encode(text))
raw_log_lines = [f"[2026-07-09T10:{i:02d}:00Z] step-{i} INFO ok" for i in range(400)]
raw_log_text = "\n".join(raw_log_lines)
trimmed_log_text = "\n".join(raw_log_lines[-50:])
print(f"Raw log tokens : {count_tokens(raw_log_text):,}")
print(f"Trimmed log tokens: {count_tokens(trimmed_log_text):,}")
Raw log tokens : 8,460
Trimmed log tokens: 615
Giảm 93% cho riêng kết quả của get_deployment_logs — và đây là một trường hợp điển hình, vì log CI/CD là một trong những loại kết quả tool "bùng nổ token" nhiều nhất trong thực tế. Với list_pull_requests, việc trim từ đối tượng PR đầy đủ (khoảng 15-18 field, bao gồm object user lồng nhau, thông tin base/head) xuống 7 field cốt lõi thường giảm khoảng 60-70% token trên mỗi PR.
Mẹo
Trim không có nghĩa là cắt bỏ vĩnh viễn thông tin — hãy luôn để lại một "lối vào" để agent lấy chi tiết đầy đủ khi thực sự cần, ví dụ giữ tool get_pull_request (chi tiết một PR) tách biệt với list_pull_requests (danh sách rút gọn) như trong bài lab này. Nguyên tắc là: liệt kê rẻ, chi tiết đắt hơn nhưng chỉ trả tiền khi thật sự cần.
Bước 4: Cấu Hình MCP Client Để Batching
Ba bước trước tối ưu nội dung mỗi round-trip (round-trip — một lượt gọi qua lại giữa agent và LLM). Bước này tối ưu số lượng round-trip, bằng cách khuyến khích model gộp nhiều tool call độc lập vào cùng một turn, và thực thi chúng song song ở phía client.
Tạo mcp_batching_client.py. Phần lõi là một hàm chạy đồng thời (concurrently) danh sách tool call mà model yêu cầu trong một response, dùng asyncio.gather:
import asyncio
import json
async def call_tool_via_mcp(tool_name: str, arguments: dict) -> dict:
"""Placeholder for the real MCP call — in a full implementation this
sends a tools/call request over the MCP client session (stdio or SSE
transport) and awaits the response."""
# NOTE: replace with a real ClientSession.call_tool(...) in production.
await asyncio.sleep(0.05) # simulate network/IO latency
return {"tool_name": tool_name, "arguments": arguments, "result": "mock-result"}
async def execute_tool_calls(tool_calls: list[dict]) -> list[dict]:
"""Execute all tool calls requested by the model in a single assistant
turn concurrently, returning results tagged by tool_use_id so they can
be mapped back correctly."""
async def run_one(call: dict) -> dict:
try:
result = await call_tool_via_mcp(call["name"], call["arguments"])
return {"tool_use_id": call["id"], "content": result, "is_error": False}
except Exception as exc: # defensive: one failed call must not block the batch
return {"tool_use_id": call["id"], "content": str(exc), "is_error": True}
return await asyncio.gather(*(run_one(c) for c in tool_calls))
SYSTEM_PROMPT_BATCHING_HINT = (
"When you need information from more than one independent tool "
"(for example checking CI status AND listing recent deployments), "
"call all of them in the same turn instead of one at a time. "
"Only call tools sequentially when a later call's input truly "
"depends on the output of an earlier call."
)
async def main() -> None:
# Example batch as the model would emit it in one assistant turn.
tool_calls = [
{"id": "call_1", "name": "get_ci_status", "arguments": {"repo": "billing-service", "ref": "main"}},
{"id": "call_2", "name": "list_deployments", "arguments": {"repo": "billing-service", "environment": "production"}},
]
results = await execute_tool_calls(tool_calls)
print(json.dumps(results, indent=2))
if __name__ == "__main__":
asyncio.run(main())
python mcp_batching_client.py
Điểm cần nhớ: bản thân việc thực thi song song ở client không tự động xảy ra trừ khi model chủ động phát ra nhiều tool_use block trong cùng một turn — và model chỉ làm vậy khi được gợi ý rõ trong system prompt (SYSTEM_PROMPT_BATCHING_HINT phía trên). Nếu bỏ qua bước gợi ý này, code batching ở client vẫn đúng nhưng vô dụng, vì model vẫn gọi tool tuần tự từng cái một.
Với hai tool call độc lập như ví dụ trên, batching biến 2 round-trip thành 1 — tiết kiệm toàn bộ chi phí gửi lại context (tool definitions đã filter + lịch sử hội thoại tích lũy) của round-trip thứ hai. Trong một session điển hình của DevOps Toolkit — nơi agent thường cần kiểm tra 2-4 nguồn thông tin độc lập trước khi trả lời — batching thường giảm số round-trip từ 8-9 xuống 3-4.
Mẹo
Đừng chỉ thêm gợi ý batching vào system prompt rồi tin tưởng mù quáng — hãy log số lượng tool_use block trung bình mỗi assistant turn trước và sau khi thêm SYSTEM_PROMPT_BATCHING_HINT. Nếu con số này không tăng đáng kể, model của bạn (hoặc phiên bản model) có thể cần ví dụ cụ thể hơn (few-shot) trong system prompt thay vì chỉ dẫn trừu tượng.
Bước 5: Đo Lường Cấu Hình Tối Ưu Cuối Cùng
Đến đây, ta có đủ bốn thành phần: schema đã nén (bước 1), tool đã filter theo intent (bước 2), kết quả đã trim (bước 3), và round-trip đã batch (bước 4). Bước cuối là mô phỏng một session thực tế để đo tổng hiệu quả cộng gộp — vì các kỹ thuật này không cộng tuyến tính, chúng nhân với nhau.
Viết final_measurement.py, mô phỏng một session gồm 6 lượt hỏi của người dùng (một buổi làm việc điển hình: kiểm tra PR, xem CI, xem log lỗi deploy, tạo issue, v.v.), với các thông số đo được từ các bước trước:
BASELINE = {
"tool_defs_tokens_per_round_trip": 3820, # 12 verbose tools, sent every round-trip
"round_trips": 14, # no batching -> sequential tool calls
"avg_result_tokens_per_call": 1450, # untrimmed results (e.g. full PR objects, full logs)
"total_tool_calls": 16,
}
OPTIMIZED = {
"tool_defs_tokens_per_round_trip": 640, # compressed schema (Step 1) + filtered to ~4 tools (Step 2)
"round_trips": 7, # batching (Step 4) roughly halves round-trips
"avg_result_tokens_per_call": 205, # trimmed results (Step 3)
"total_tool_calls": 16, # same underlying work, cheaper per call
}
def session_cost(cfg: dict) -> dict:
tool_def_overhead = cfg["tool_defs_tokens_per_round_trip"] * cfg["round_trips"]
result_overhead = cfg["avg_result_tokens_per_call"] * cfg["total_tool_calls"]
return {
"tool_def_overhead": tool_def_overhead,
"result_overhead": result_overhead,
"total": tool_def_overhead + result_overhead,
}
def main() -> None:
baseline_cost = session_cost(BASELINE)
optimized_cost = session_cost(OPTIMIZED)
savings_pct = (1 - optimized_cost["total"] / baseline_cost["total"]) * 100
print("=== Baseline session ===")
print(f" Tool definitions overhead: {baseline_cost['tool_def_overhead']:,} tokens")
print(f" Tool results overhead : {baseline_cost['result_overhead']:,} tokens")
print(f" Total : {baseline_cost['total']:,} tokens")
print("=== Optimized session ===")
print(f" Tool definitions overhead: {optimized_cost['tool_def_overhead']:,} tokens")
print(f" Tool results overhead : {optimized_cost['result_overhead']:,} tokens")
print(f" Total : {optimized_cost['total']:,} tokens")
print(f"\nOverall savings: {savings_pct:.1f}%")
if __name__ == "__main__":
main()
python final_measurement.py
Kết quả mẫu:
=== Baseline session ===
Tool definitions overhead: 53,480 tokens
Tool results overhead : 23,200 tokens
Total : 76,680 tokens
=== Optimized session ===
Tool definitions overhead: 4,480 tokens
Tool results overhead : 3,280 tokens
Total : 7,760 tokens
Overall savings: 89.9%
Con số 89.9% có thể trông "đẹp quá mức" — nhưng nó phản ánh đúng bản chất toán học của việc kết hợp nhiều tối ưu độc lập: nén schema giảm ~57%, filtering giảm tiếp ~60% trên phần còn lại, trimming giảm ~85% trên kết quả, và batching giảm ~50% số round-trip. Các hệ số này nhân với nhau chứ không cộng, nên tổng hợp lại tạo ra một con số lớn. Trong thực tế production, mức baseline của bạn có thể chưa "tệ" như ví dụ minh họa này (nếu tool definitions đã tương đối gọn từ đầu), nên hãy kỳ vọng mức tiết kiệm thực tế nằm trong khoảng 40-70% là hợp lý và bền vững — và luôn đối chiếu với baseline thật của hệ thống bạn, không dùng số liệu minh họa trong bài lab này làm chuẩn.
Mẹo
Luôn tách riêng "tool definitions overhead" và "tool results overhead" trong báo cáo đo lường, như bảng trên. Hai loại overhead này cần hai nhóm kỹ thuật khác nhau để xử lý (nén/filter cho definitions, trim cho results) — gộp chung một con số duy nhất khiến bạn khó biết nên đầu tư công sức tối ưu vào đâu tiếp theo.
Mở Rộng Bài Lab: Thêm Thử Thách Tối Ưu
Bài lab trên đã đi qua bốn kỹ thuật cốt lõi, nhưng còn nhiều hướng mở rộng đáng để bạn tự thử nếu muốn đào sâu hơn:
- Đo bằng tokenizer thật của model bạn dùng: thay
tiktokenbằng endpointcount_tokenscủa Anthropic API (hoặc endpoint tương đương của provider bạn dùng) để có số liệu tuyệt đối chính xác, không chỉ là ước lượng tương đối. - Thêm prompt caching cho tool definitions: nếu provider hỗ trợ prompt caching, đánh dấu phần tool definitions ít thay đổi làm cache breakpoint — kết hợp với dynamic filtering, bạn có thể cache riêng từng nhóm tool theo intent.
- Tool retrieval bằng embedding cho toolset lớn: khi số tool vượt quá 30-50 (vượt xa quy mô 12 tool của bài lab), keyword classifier không còn đủ chính xác — hãy thử semantic search bằng embedding trên description tool để chọn tool liên quan nhất cho mỗi request.
- Phân trang (pagination) thay vì trim cứng: với
get_deployment_logs, thay vì luôn trả 50 dòng cuối, thử thêm tham sốoffset/cursorđể agent có thể "cuộn" qua log khi 50 dòng đầu chưa đủ, tránh việc agent phải gọi lại toàn bộ log thô khi cần thêm chi tiết. - Kết nối với MCP server thực tế: áp lại toàn bộ quy trình đo/tối ưu này trên một MCP server có sẵn (ví dụ GitHub MCP server, Jira/Linear MCP server) để xem mức tiết kiệm thực tế trên toolset production, không phải toolset giả lập.
- Theo dõi tỷ lệ thành công task (task success rate) song song với token: mọi kỹ thuật trong bài lab đều có thể làm giảm token trong khi vô tình làm agent chọn sai tool hoặc thiếu ngữ cảnh — hãy dựng một bộ test case (ít nhất 15-20 tình huống) để chạy trước/sau mỗi bước tối ưu, đảm bảo tỷ lệ hoàn thành task không giảm khi token giảm.
Mẹo
Chọn tối đa 1-2 thử thách mở rộng để làm sâu, thay vì làm dàn trải cả năm hướng. Semantic tool retrieval và prompt caching là hai hướng có ROI cao nhất nếu bạn đang vận hành MCP server với toolset lớn (>20 tool) trong production thực tế.
Tổng Kết
Bài lab này đã đưa bạn qua toàn bộ vòng đời tối ưu một MCP server, từ việc dựng baseline đo được bằng số liệu cụ thể, đến áp lần lượt bốn kỹ thuật: nén tool schema (giảm ~57% token định nghĩa tool), dynamic tool filtering theo intent (giảm tiếp phần lớn số tool gửi mỗi request), trim kết quả tool handler (giảm tới hơn 90% cho các loại kết quả cồng kềnh như log), và batching tool call để giảm số round-trip. Điểm mấu chốt xuyên suốt không phải là ghi nhớ từng đoạn code, mà là quy trình: đo trước — tối ưu một thứ — đo lại — so sánh, lặp lại cho từng lớp overhead (tool definitions, kết quả tool, số round-trip) một cách tách biệt và có kiểm chứng.
Khi mang các kỹ thuật này vào một MCP server thực tế của bạn, hãy nhớ ba nguyên tắc: (1) luôn đo bằng tokenizer sát nhất với model bạn dùng trong production, không chỉ dựa vào ước lượng của tiktoken; (2) mọi tối ưu giảm token đều phải đi kèm kiểm tra tỷ lệ thành công task, vì token thấp mà agent làm sai việc thì không có ý nghĩa gì; và (3) các kỹ thuật này nhân với nhau, không cộng — nên đầu tư đều vào cả bốn lớp (schema, filtering, trimming, batching) thường mang lại hiệu quả tổng thể lớn hơn nhiều so với việc chỉ dồn toàn lực vào một lớp duy nhất.