在前幾篇文章中，我們學習了 Claude Agent SDK 的基礎概念、Guardrails 護欄機制，以及實戰專案的開發方式。本篇將進入企業應用的核心主題：如何將 Claude Agent SDK 整合進現有的企業工具鏈，讓 AI Agent 真正成為開發、維運、業務流程的一部分。
企業工具鏈整合的挑戰
企業環境與個人開發環境最大的差異在於複雜度：一個大型工程團隊可能同時使用 GitHub 管理程式碼、Jira 追蹤 issue、Slack 溝通協作、Sentry 監控錯誤、PostgreSQL 儲存資料，再加上各種內部系統。要讓 AI Agent 有效運作，必須解決以下幾個挑戰：
- 多系統整合：Agent 需要同時存取多個工具和資料來源
- 身份驗證：每個系統有不同的 API Key 和 OAuth 流程
- 權限控制：不同角色的 Agent 應有不同的操作權限
- 審計追蹤：企業需要記錄 Agent 的所有操作
- 錯誤處理：生產環境不容許 Agent 無控制地失敗
Claude Agent SDK 透過 MCP（Model Context Protocol）架構和 Hooks 機制，提供了解決這些挑戰的完整方案。
MCP Server 作為企業整合橋樑
MCP（Model Context Protocol）是連接 Agent 與外部系統的標準協議。截至 2026 年，已有超過 300 個官方和社群維護的 MCP Server，涵蓋主流企業工具。以下是常見的企業工具整合清單：
| 工具類別 | 常見 MCP Server | 主要功能 |
|---|---|---|
| 版本控制 | @modelcontextprotocol/server-github | PR 管理、Issue 操作、程式碼搜尋 |
| 專案管理 | @linear/mcp、@atlassian/jira-mcp | Task 追蹤、Sprint 管理、工作分配 |
| 溝通協作 | @slack/mcp-server | 發送訊息、查詢頻道、監控通知 |
| 錯誤監控 | @sentry/mcp-server | 查詢錯誤、分析堆疊追蹤 |
| 資料庫 | @modelcontextprotocol/server-postgres | 查詢、分析資料庫 |
| 文件管理 | @google/drive-mcp、@notion/mcp | 文件搜尋、建立與更新 |
| CI/CD | 自訂 MCP（見下方範例） | 觸發 Pipeline、查詢建置狀態 |
多 MCP Server 配置範例
以下範例展示如何在 Agent 中同時連接 GitHub、Sentry 和 Slack，實現一個完整的「Bug 偵測與通報」工作流：
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
async def bug_detection_workflow(pr_number: int):
async for message in query(
prompt=(
f"請執行以下工作流程：\n"
f"1. 使用 GitHub MCP 取得 PR #{pr_number} 的變更內容\n"
f"2. 使用 Sentry MCP 查詢最近 24 小時內與這些檔案相關的錯誤\n"
f"3. 分析 PR 變更是否可能引發已知問題\n"
f"4. 將分析結果發送到 Slack #code-review 頻道"
),
options=ClaudeAgentOptions(
allowed_tools=[
"mcp__github__get_pull_request",
"mcp__github__list_pull_request_files",
"mcp__sentry__get_issues",
"mcp__sentry__get_events",
"mcp__slack__post_message",
],
mcp_servers={
"github": {
"command": "npx",
"args": ["@modelcontextprotocol/server-github"],
"env": {
"GITHUB_TOKEN": "ghp_your_github_token"
}
},
"sentry": {
"command": "npx",
"args": ["@sentry/mcp-server"],
"env": {
"SENTRY_AUTH_TOKEN": "sntrys_your_token",
"SENTRY_ORG": "your-org"
}
},
"slack": {
"command": "npx",
"args": ["@slack/mcp-server"],
"env": {
"SLACK_BOT_TOKEN": "xoxb-your-slack-token"
}
}
},
system_prompt="你是一位資深 Code Reviewer，擅長識別潛在的 Bug 和效能問題。"
),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(bug_detection_workflow(pr_number=1234))自訂 In-Process 工具
除了使用現有的 MCP Server，你也可以建立執行在同一個 Python 程序內（In-Process）的自訂工具。這種方式特別適合需要存取內部資料庫、公司私有 API，或不適合暴露為獨立程序的邏輯。
import asyncio
import json
from claude_agent_sdk import query, ClaudeAgentOptions
from mcp.server import Server
import mcp.types as types
# 建立一個 In-Process MCP Server
app = Server("internal-tools")
@app.list_tools()
async def list_tools():
return [
types.Tool(
name="get_employee_info",
description="查詢員工資訊，包括部門、職位、聯絡方式",
inputSchema={
"type": "object",
"properties": {
"employee_id": {
"type": "string",
"description": "員工編號（例如 EMP-001）"
}
},
"required": ["employee_id"]
}
),
types.Tool(
name="submit_deployment_request",
description="提交部署申請，需要主管核准",
inputSchema={
"type": "object",
"properties": {
"service": {"type": "string"},
"version": {"type": "string"},
"environment": {
"type": "string",
"enum": ["staging", "production"]
},
"reason": {"type": "string"}
},
"required": ["service", "version", "environment", "reason"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "get_employee_info":
emp_id = arguments["employee_id"]
data = {
"EMP-001": {"name": "王小明", "dept": "工程部", "title": "資深工程師"},
"EMP-002": {"name": "李小華", "dept": "產品部", "title": "產品經理"},
}
result = data.get(emp_id, {"error": "找不到此員工"})
return [types.TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]
elif name == "submit_deployment_request":
ticket_id = "DEP-" + str(hash(str(arguments)) % 10000).zfill(4)
return [types.TextContent(
type="text",
text=json.dumps({
"ticket_id": ticket_id,
"status": "pending_approval",
"message": "部署申請已提交，票號：" + ticket_id + "，請等待主管核准"
}, ensure_ascii=False)
)]
async def run_with_internal_tools():
async for message in query(
prompt="查詢員工 EMP-001 的資訊，並替他提交一個將 api-service v2.1.0 部署到 staging 的申請",
options=ClaudeAgentOptions(
allowed_tools=[
"mcp__internal-tools__get_employee_info",
"mcp__internal-tools__submit_deployment_request"
],
mcp_servers={
"internal-tools": {
"server": app,
}
}
),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(run_with_internal_tools())企業級工具鏈架構設計
在設計企業級 Agent 工具鏈時，建議採用分層架構，讓不同層級的 Agent 各司其職：
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition
# 第一層：專門化子 Agent
# 程式碼分析子 Agent（只能讀取，不能修改）
code_analyst = AgentDefinition(
description="分析程式碼品質、找出潛在問題與技術債",
prompt=(
"你是程式碼品質分析師。\n"
"- 使用 Glob 和 Read 掃描程式碼\n"
"- 識別重複程式碼、複雜度過高的函式\n"
"- 找出潛在的安全問題（SQL Injection、XSS 等）\n"
"- 輸出結構化的分析報告"
),
tools=["Read", "Glob", "Grep"]
)
# 部署協調子 Agent（可以操作 CI/CD 系統）
deployment_coordinator = AgentDefinition(
description="協調部署流程，包括觸發 Pipeline 和監控建置狀態",
prompt=(
"你是部署協調員。\n"
"- 使用 GitHub MCP 建立 Release Tag\n"
"- 觸發 CI/CD Pipeline\n"
"- 監控建置和測試結果\n"
"- 若有失敗，整理錯誤摘要"
),
tools=["mcp__github__create_tag", "mcp__github__get_workflow_run", "Bash"]
)
# 通報子 Agent（只能發送訊息，不能執行操作）
reporter = AgentDefinition(
description="整理並傳送通知到適當的溝通頻道",
prompt=(
"你是通報專員。\n"
"- 將技術報告轉換為清晰易讀的摘要\n"
"- 依據嚴重程度選擇通報管道（Slack、Email）\n"
"- 確保通知包含必要的行動項目"
),
tools=["mcp__slack__post_message"]
)
# 第二層：主 Agent（協調者）
async def enterprise_release_pipeline(repo: str, version: str):
async for message in query(
prompt=(
"執行 " + repo + " v" + version + " 的完整發布流程：\n"
"1. 使用 code-analyst 對最新的變更進行品質分析\n"
"2. 若分析結果無重大問題，使用 deployment-coordinator 執行部署\n"
"3. 使用 reporter 將結果通報到 Slack #deployments 頻道\n"
"\n若任何步驟失敗，立即停止並通報問題。"
),
options=ClaudeAgentOptions(
system_prompt="你是發布協調主管，負責確保每次發布都安全、有序地進行。",
allowed_tools=["Agent", "mcp__github__list_commits"],
mcp_servers={
"github": {
"command": "npx",
"args": ["@modelcontextprotocol/server-github"],
"env": {"GITHUB_TOKEN": "ghp_your_token"}
},
"slack": {
"command": "npx",
"args": ["@slack/mcp-server"],
"env": {"SLACK_BOT_TOKEN": "xoxb-your_token"}
}
},
agents={
"code-analyst": code_analyst,
"deployment-coordinator": deployment_coordinator,
"reporter": reporter
}
),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(enterprise_release_pipeline("api-service", "2.1.0"))環境變數與密鑰管理
在企業環境中，API 密鑰的管理至關重要。絕對不要將密鑰硬編碼在程式中，而應使用環境變數或密鑰管理服務：
import os
import asyncio
from dotenv import load_dotenv
from claude_agent_sdk import query, ClaudeAgentOptions
load_dotenv()
def build_mcp_servers():
servers = {}
if github_token := os.getenv("GITHUB_TOKEN"):
servers["github"] = {
"command": "npx",
"args": ["@modelcontextprotocol/server-github"],
"env": {"GITHUB_TOKEN": github_token}
}
if slack_token := os.getenv("SLACK_BOT_TOKEN"):
servers["slack"] = {
"command": "npx",
"args": ["@slack/mcp-server"],
"env": {"SLACK_BOT_TOKEN": slack_token}
}
if (sentry_token := os.getenv("SENTRY_AUTH_TOKEN")) and \
(sentry_org := os.getenv("SENTRY_ORG")):
servers["sentry"] = {
"command": "npx",
"args": ["@sentry/mcp-server"],
"env": {
"SENTRY_AUTH_TOKEN": sentry_token,
"SENTRY_ORG": sentry_org
}
}
if pg_url := os.getenv("DATABASE_URL"):
servers["postgres"] = {
"command": "npx",
"args": ["@modelcontextprotocol/server-postgres", pg_url]
}
return servers
async def run_enterprise_agent(task: str):
mcp_servers = build_mcp_servers()
if not mcp_servers:
raise ValueError("至少需要設定一個 MCP Server 的環境變數")
allowed_tools = []
if "github" in mcp_servers:
allowed_tools.extend(["mcp__github__search_code", "mcp__github__get_issue"])
if "slack" in mcp_servers:
allowed_tools.append("mcp__slack__post_message")
if "sentry" in mcp_servers:
allowed_tools.append("mcp__sentry__get_issues")
if "postgres" in mcp_servers:
allowed_tools.append("mcp__postgres__query")
async for message in query(
prompt=task,
options=ClaudeAgentOptions(
mcp_servers=mcp_servers,
allowed_tools=allowed_tools
),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(run_enterprise_agent("分析最近一週的 Sentry 錯誤，並整理出最常見的 5 個問題"))Hooks 實作企業級稽核日誌
在企業環境中，所有 Agent 操作都必須留下稽核記錄。使用 Hooks 機制可以輕鬆實現：
import asyncio
import json
import logging
import os
from datetime import datetime
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher
logging.basicConfig(
level=logging.INFO,
format="%(message)s",
handlers=[
logging.FileHandler("agent_audit.jsonl"),
logging.StreamHandler()
]
)
logger = logging.getLogger("agent_audit")
def log_audit_event(event_type: str, data: dict):
event = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"event_type": event_type,
**data
}
logger.info(json.dumps(event, ensure_ascii=False))
async def on_tool_use_start(input_data, tool_use_id, context):
log_audit_event("TOOL_USE_START", {
"tool_use_id": tool_use_id,
"tool_name": input_data.get("tool_name"),
"tool_input": input_data.get("tool_input"),
})
return {}
async def on_tool_use_end(input_data, tool_use_id, context):
log_audit_event("TOOL_USE_END", {
"tool_use_id": tool_use_id,
"tool_name": input_data.get("tool_name"),
"success": True,
})
return {}
async def block_dangerous_operations(input_data, tool_use_id, context):
tool_input = input_data.get("tool_input", {})
query_str = tool_input.get("query", "").upper()
dangerous_keywords = ["DROP TABLE", "DELETE FROM", "TRUNCATE", "DROP DATABASE"]
for keyword in dangerous_keywords:
if keyword in query_str:
log_audit_event("TOOL_BLOCKED", {
"tool_use_id": tool_use_id,
"reason": "包含危險操作：" + keyword,
"query": query_str
})
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": "資安政策禁止執行 " + keyword + " 操作"
}
}
return {}
async def run_audited_agent(task: str, agent_id: str = "default"):
log_audit_event("AGENT_START", {
"agent_id": agent_id,
"task": task
})
try:
async for message in query(
prompt=task,
options=ClaudeAgentOptions(
allowed_tools=[
"Read", "Bash",
"mcp__postgres__query",
"mcp__github__search_issues"
],
mcp_servers={
"postgres": {
"command": "npx",
"args": [
"@modelcontextprotocol/server-postgres",
os.getenv("DATABASE_URL")
]
}
},
hooks={
"PreToolUse": [
HookMatcher(
matcher="mcp__postgres__query",
hooks=[block_dangerous_operations]
),
HookMatcher(hooks=[on_tool_use_start])
],
"PostToolUse": [
HookMatcher(hooks=[on_tool_use_end])
]
}
),
):
if hasattr(message, "result"):
log_audit_event("AGENT_COMPLETE", {
"agent_id": agent_id,
"success": True
})
return message.result
except Exception as e:
log_audit_event("AGENT_ERROR", {
"agent_id": agent_id,
"error": str(e)
})
raise
asyncio.run(run_audited_agent("查詢過去 30 天內新增的使用者數量", agent_id="weekly-report-001"))TypeScript 版本的企業整合範例
如果你的技術棧以 Node.js 為主，以下是 TypeScript 版本的 GitHub + Linear 整合，用於自動化 Issue 處理流程：
import { query, HookCallback, PreToolUseHookInput } from "@anthropic-ai/claude-agent-sdk";
import * as fs from "fs/promises";
// 稽核日誌 Hook
const auditLogger: HookCallback = async (input) => {
const preInput = input as PreToolUseHookInput;
const logEntry = {
timestamp: new Date().toISOString(),
event: "tool_use",
tool: preInput.tool_name,
input: preInput.tool_input
};
await fs.appendFile("audit.jsonl", JSON.stringify(logEntry) + "\n");
return {};
};
// 主工作流：GitHub Issue 轉 Linear Task
async function processGithubIssueToLinear(issueNumber: number) {
const results: string[] = [];
for await (const message of query({
prompt: (
"請執行以下 Issue 處理流程：\n" +
"1. 從 GitHub 取得 Issue #" + issueNumber + " 的完整內容（標題、描述、標籤）\n" +
"2. 分析 Issue 的類型（Bug、Feature Request、Documentation 等）\n" +
"3. 在 Linear 建立對應的 Task，包含：\n" +
" - 適當的 Priority（依嚴重程度）\n" +
" - 對應的 Team（依 Issue 標籤判斷）\n" +
" - 從 GitHub Issue 連結過來的描述\n" +
"4. 在 GitHub Issue 留言，告知已建立 Linear Task 並附上連結"
),
options: {
systemPrompt: "你是工程團隊的 Triage 助手，負責將 GitHub Issues 正確分配到 Linear。",
allowedTools: [
"mcp__github__get_issue",
"mcp__github__create_issue_comment",
"mcp__linear__create_issue",
"mcp__linear__get_teams"
],
mcpServers: {
github: {
command: "npx",
args: ["@modelcontextprotocol/server-github"],
env: { GITHUB_TOKEN: process.env.GITHUB_TOKEN! }
},
linear: {
command: "npx",
args: ["@linear/mcp"],
env: { LINEAR_API_KEY: process.env.LINEAR_API_KEY! }
}
},
hooks: {
PreToolUse: [{ hooks: [auditLogger] }]
}
}
})) {
if ("result" in message && message.result) {
results.push(message.result);
}
}
return results[results.length - 1];
}
processGithubIssueToLinear(1234)
.then(result => console.log("完成：", result))
.catch(console.error);雲端平台整合：Bedrock 與 Vertex AI
對於需要在特定雲端環境運行的企業，Claude Agent SDK 支援透過 Amazon Bedrock、Google Vertex AI 和 Microsoft Azure AI Foundry 存取，無需直接呼叫 Anthropic API：
| 雲端平台 | 環境變數設定 | 適用情境 |
|---|---|---|
| Anthropic API（直接） | ANTHROPIC_API_KEY | 快速開發、一般企業 |
| Amazon Bedrock | CLAUDE_CODE_USE_BEDROCK=1 + AWS 憑證 | AWS 生態系、需要 VPC 隔離 |
| Google Vertex AI | CLAUDE_CODE_USE_VERTEX=1 + GCP 憑證 | GCP 生態系、跨境資料合規 |
| Microsoft Azure AI Foundry | CLAUDE_CODE_USE_FOUNDRY=1 + Azure 憑證 | Microsoft 企業客戶、歐盟合規需求 |
import os
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions
# 使用 Amazon Bedrock（只需設定環境變數，程式碼完全相同）
os.environ["CLAUDE_CODE_USE_BEDROCK"] = "1"
os.environ["AWS_REGION"] = "us-east-1"
# AWS_ACCESS_KEY_ID 和 AWS_SECRET_ACCESS_KEY 應從 IAM Role 或 ~/.aws/credentials 取得
async def run_on_bedrock():
async for message in query(
prompt="分析這個專案的安全性設定",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Glob", "Grep"],
),
):
if hasattr(message, "result"):
print(message.result)
asyncio.run(run_on_bedrock())企業工具鏈整合最佳實踐
1. 最小權限原則
每個 Agent 只應獲得完成任務所需的最低權限。不要讓同一個 Agent 既能讀取資料庫又能發送外部通知，除非業務流程確實需要。使用 allowed_tools 精確限制工具存取範圍。
2. 密鑰不入程式碼
所有 API Token、資料庫連線字串都應透過環境變數或密鑰管理服務（如 AWS Secrets Manager、HashiCorp Vault）注入。考慮為不同環境（開發、測試、生產）設立不同的密鑰，並定期輪替。
3. 完整的稽核日誌
使用 Hooks 記錄所有工具呼叫。稽核日誌應包含：時間戳記、Agent ID、工具名稱、輸入參數、執行結果。建議使用結構化 JSON 格式，方便後續查詢和分析。
4. 幂等性設計
在設計 Agent 工作流時，確保操作可以安全地重複執行。例如，建立 GitHub Issue 前先檢查是否已存在、發送 Slack 通知前確認未重複發送。這在 Agent 遇到錯誤需要重試時格外重要。
5. 錯誤處理與降級策略
為每個 MCP Server 設計降級（Fallback）策略。例如，若 Slack MCP 無法連線，Agent 應該能夠改用 Email 通知或記錄到本地日誌，而不是直接失敗中止。
總結
企業級 Agent 工具鏈整合的核心在於：以 MCP Server 作為橋樑連接外部系統、使用 Hooks 實現稽核與安全控制、透過分層 Agent 架構處理複雜工作流、並在雲端平台層級選擇最適合企業合規需求的部署方式。
掌握了這些技術，你就能將 Claude Agent SDK 真正融入企業的 DevOps 流程、客服系統、資料分析管道等各種場景，打造出能自主運作、有稽核記錄、又符合資安要求的企業級 AI 工具鏈。
在接下來的系列文章中，我們將進入 Anthropic Messages API 的深度應用，探索如何直接透過 API 打造更靈活的 AI 應用程式。
