Strands Agents Harness SDK 完整教學

1# 使用 pip 安裝核心 SDK
2pip install strands-agents
3
4# 安裝官方工具集（選用，但強烈建議）
5pip install strands-agents-tools
6
7# 如果使用 uv（推薦）
8uv pip install strands-agents strands-agents-tools

 1# Anthropic 直接 API
 2pip install strands-agents[anthropic]
 3
 4# OpenAI
 5pip install strands-agents[openai]
 6
 7# Ollama（本地模型）
 8pip install strands-agents[ollama]
 9
10# LiteLLM（統一代理）
11pip install strands-agents[litellm]

1npm install @strands-agents/sdk

 1# 方法 1：設定環境變數
 2export AWS_ACCESS_KEY_ID="your-access-key"
 3export AWS_SECRET_ACCESS_KEY="your-secret-key"
 4export AWS_DEFAULT_REGION="us-east-1"
 5
 6# 方法 2：使用 AWS CLI 設定
 7aws configure
 8
 9# 方法 3：使用 AWS SSO
10aws sso login --profile your-profile

1export ANTHROPIC_API_KEY="sk-ant-..."

1from strands import Agent
2from strands.models.anthropic import AnthropicModel
3
4model = AnthropicModel(model_id="claude-sonnet-4-20250514")
5agent = Agent(model=model)

1export OPENAI_API_KEY="sk-..."

1from strands import Agent
2from strands.models.openai import OpenAIModel
3
4model = OpenAIModel(model_id="gpt-4o")
5agent = Agent(model=model)

1# 先啟動 Ollama 並拉取模型
2ollama pull llama3.1
3
4# 使用
5from strands import Agent
6from strands.models.ollama import OllamaModel
7
8model = OllamaModel(host="http://localhost:11434", model_id="llama3.1")
9agent = Agent(model=model)

 1from strands import Agent
 2from strands_tools import calculator
 3
 4# 建立 Agent，配備計算機工具
 5# Agent 預設使用 Amazon Bedrock 作為 Model Provider
 6agent = Agent(tools=[calculator])
 7
 8# 呼叫 Agent —— 就像呼叫一個函式一樣直覺
 9result = agent("1764 的平方根是多少？")
10# Agent 會自動決定是否使用 calculator 工具
11# 最終回傳 AgentResult 物件

 1from strands import Agent, tool
 2
 3@tool
 4def get_weather(city: str, unit: str = "celsius") -> dict:
 5    """取得指定城市的天氣資訊。
 6    
 7    Args:
 8        city: 城市名稱，例如 "Taipei"。
 9        unit: 溫度單位，可選 "celsius" 或 "fahrenheit"（預設 celsius）。
10    
11    Returns:
12        包含天氣資訊的字典。
13    """
14    # 實際應用中會呼叫天氣 API
15    weather_data = {
16        "city": city,
17        "temperature": 28,
18        "unit": unit,
19        "condition": "partly cloudy"
20    }
21    return {
22        "status": "success",
23        "content": [{"text": f"{city} 目前 {weather_data['temperature']}°C，{weather_data['condition']}"}]
24    }
25
26# 建立 Agent 並載入自訂工具
27agent = Agent(tools=[get_weather])
28result = agent("台北現在天氣如何？")

 1from strands import Agent
 2from strands.models.anthropic import AnthropicModel
 3
 4# 建立專門做翻譯的子 Agent
 5translator = Agent(
 6    model=AnthropicModel(model_id="claude-sonnet-4-20250514"),
 7    system_prompt="你是專業翻譯員。將使用者輸入翻譯成英文。只回傳翻譯結果，不加解釋。",
 8    name="translator",
 9    description="將中文翻譯成英文的翻譯代理人"
10)
11
12# 建立專門做摘要的子 Agent
13summarizer = Agent(
14    system_prompt="你是摘要專家。將輸入內容濃縮為 3 個重點。",
15    name="summarizer",
16    description="將長文濃縮為重點摘要的代理人"
17)
18
19# 建立指揮 Agent，將兩個子 Agent 作為工具
20# 當 Agent 物件被放入 tools 清單時，SDK 會自動呼叫 agent.as_tool()
21# 將其包裝為 _AgentAsTool，模型可透過 tool_use 呼叫它
22orchestrator = Agent(
23    system_prompt="你是協調者。根據使用者需求，分派工作給適當的助手。",
24    tools=[translator, summarizer]  # Agent 直接作為 tool
25)
26
27# 指揮 Agent 會根據需求自動選擇呼叫哪個子 Agent
28result = orchestrator("請把以下中文摘要後翻譯成英文：台灣位於東亞...")

 1from strands import Agent, InterventionHandler
 2from strands.hooks import BeforeToolCallEvent, AfterToolCallEvent
 3from strands.interventions.actions import Proceed, Deny, Guide
 4
 5import logging
 6
 7# 方法 1：使用 Intervention Handler 做 Guardrail (護欄)
 8class SafetyGuard(InterventionHandler):
 9    """在工具執行前檢查安全性。"""
10    
11    name = "safety-guard"  # 每個 handler 必須有唯一名稱
12    
13    def before_tool_call(self, event: BeforeToolCallEvent, **kwargs):
14        """在工具被呼叫前攔截，檢查是否允許執行。"""
15        tool_name = event.tool.tool_name
16        
17        # 禁止使用某些危險工具
18        blocked_tools = ["shell_command", "file_delete"]
19        if tool_name in blocked_tools:
20            # Deny 會立即終止這個工具呼叫，回傳拒絕原因給模型
21            return Deny(reason=f"工具 {tool_name} 已被安全政策封鎖")
22        
23        # Guide 會在不阻止執行的情況下，向模型注入額外指引
24        if tool_name == "web_search":
25            return Guide(instructions="搜尋時請避免存取內部網域")
26        
27        # Proceed 表示允許繼續執行
28        return Proceed()
29
30# 方法 2：使用 Hook Callback 做日誌記錄
31def log_tool_usage(event: AfterToolCallEvent):
32    """在工具執行後記錄使用情況。"""
33    logging.info(
34        f"工具 {event.tool.tool_name} 執行完成，"
35        f"耗時 {event.duration_ms:.1f}ms"
36    )
37
38# 建立 Agent，同時註冊 intervention 和 hook
39agent = Agent(
40    interventions=[SafetyGuard()],    # Intervention 按註冊順序評估
41    hooks=[log_tool_usage],            # Hook 不影響控制流
42    tools=[get_weather]
43)

 1from strands.plugins import Plugin, hook
 2from strands.hooks import BeforeModelCallEvent
 3from strands import tool
 4
 5class AuditPlugin(Plugin):
 6    """審計外掛：記錄所有模型呼叫並提供查詢工具。"""
 7    
 8    name = "audit-plugin"
 9    
10    def __init__(self):
11        super().__init__()
12        self.call_log = []
13    
14    @hook
15    def on_model_call(self, event: BeforeModelCallEvent):
16        """用 @hook 裝飾的方法會自動被發現並註冊。"""
17        self.call_log.append({
18            "timestamp": "...",
19            "message_count": len(event.messages)
20        })
21    
22    @tool
23    def query_audit_log(self, query: str) -> str:
24        """查詢審計記錄。
25        
26        Args:
27            query: 查詢條件。
28        """
29        return f"共 {len(self.call_log)} 筆記錄"
30
31agent = Agent(plugins=[AuditPlugin()])

 1from pydantic import BaseModel
 2from strands import Agent
 3
 4class MovieReview(BaseModel):
 5    title: str
 6    score: float
 7    summary: str
 8    pros: list[str]
 9    cons: list[str]
10
11agent = Agent(structured_output_model=MovieReview)
12result = agent("評論電影《乘風破浪》")
13
14# result.output 是 MovieReview 型別的物件
15review = result.output
16print(f"評分：{review.score}")
17print(f"優點：{review.pros}")

 1from strands import Agent
 2from strands.session import FileSessionManager
 3
 4# 使用檔案系統做 session 持久化
 5session_mgr = FileSessionManager(session_dir="./sessions")
 6
 7agent = Agent(
 8    session_manager=session_mgr,
 9    agent_id="my-assistant"
10)
11
12# 第一次對話
13result = agent("我叫小明，請記住我的名字")
14
15# 之後的對話可以延續上下文
16# session_manager 會自動載入之前的 messages
17result = agent("我叫什麼名字？")

 1from strands import Agent
 2from strands.memory import MemoryManager
 3
 4# MemoryManager 是一個 Plugin，可以搭配多種 MemoryStore
 5memory = MemoryManager(
 6    stores=[my_memory_store],
 7    search_tool_config=True,      # 自動註冊 search_memory 工具
 8    add_tool_config=True,         # 自動註冊 add_memory 工具
 9    injection=True                # 每次模型呼叫前自動注入相關記憶
10)
11
12agent = Agent(memory_manager=memory)

 1from strands import Agent
 2from strands.sandbox import DockerSandbox
 3
 4# 在 Docker 容器中執行
 5sandbox = DockerSandbox(
 6    container="my-agent-sandbox",     # 容器名稱或 ID
 7    working_dir="/workspace",          # 工作目錄
 8    user="agent"                       # 執行用戶
 9)
10
11agent = Agent(sandbox=sandbox)
12# 現在所有 sandbox-aware 的工具都會在容器內執行

 1from strands.agent.a2a_agent import A2AAgent
 2
 3# 連接遠端 A2A Agent
 4remote_agent = A2AAgent(
 5    endpoint="https://my-agent.example.com",
 6    timeout=300
 7)
 8
 9# 像本地 Agent 一樣呼叫
10result = remote_agent("分析這份資料")
11
12# 也可以作為工具整合到其他 Agent
13orchestrator = Agent(tools=[remote_agent])

 1agent = Agent(checkpointing=True)
 2
 3result = agent("執行多步驟分析")
 4
 5# 如果 stop_reason 是 "checkpoint"，表示 Agent 在等待
 6if result.stop_reason == "checkpoint":
 7    # 檢視目前狀態，決定是否繼續
 8    checkpoint = result.checkpoint
 9    # 繼續執行
10    result = agent({"checkpointResume": {"checkpoint": checkpoint}})

1agent = Agent(context_manager="auto")

1from strands import Agent
2from strands.tools.mcp import MCPClient
3
4# 連接 MCP Server 並使用其工具
5mcp = MCPClient("npx -y @strands-agents/mcp-server")
6agent = Agent(tools=[mcp])

1# 1. 同步呼叫
2result = agent("hello")
3
4# 2. 異步串流
5async for event in agent.stream_async("hello"):
6    print(event)
7
8# 3. 結構化輸出
9result = agent.structured_output("hello", output_model=MyModel)

 1harness-sdk (核心)
 2  |
 3  +-- tools (官方工具集，pip install strands-agents-tools)
 4  |     |
 5  |     +-- shell (安全 shell 工具的底層)
 6  |
 7  +-- mcp-server (讓 AI 編輯器理解 Strands)
 8  |
 9  +-- agent-builder (基於 SDK 的示範 Agent)
10  |
11  +-- samples (更多使用範例)
12  |
13  +-- evals (品質驗證框架)
14  |
15  +-- agent-sop (SOP 工作流引擎)

 1from strands import Agent, tool, InterventionHandler
 2from strands.interventions.actions import Proceed, Deny
 3from strands.hooks import BeforeToolCallEvent
 4from strands.plugins import Plugin, hook
 5from strands.session import FileSessionManager
 6from strands.models.anthropic import AnthropicModel
 7
 8# 1. 定義查詢工具
 9@tool
10def search_knowledge_base(query: str, department: str = "all") -> dict:
11    """搜尋企業知識庫。
12    
13    Args:
14        query: 搜尋關鍵字。
15        department: 部門篩選（all / engineering / hr / finance）。
16    """
17    # 實際會接 Elasticsearch / OpenSearch
18    results = do_search(query, department)
19    return {"status": "success", "content": [{"text": str(results)}]}
20
21# 2. 定義權限檢查 Intervention
22class DepartmentAuth(InterventionHandler):
23    name = "dept-auth"
24    
25    def __init__(self, user_dept: str):
26        self.user_dept = user_dept
27    
28    def before_tool_call(self, event: BeforeToolCallEvent, **kwargs):
29        if event.tool.tool_name == "search_knowledge_base":
30            dept = event.tool_use.get("input", {}).get("department", "all")
31            if dept != "all" and dept != self.user_dept:
32                return Deny(reason=f"無權限存取 {dept} 部門的資料")
33        return Proceed()
34
35# 3. 組裝
36agent = Agent(
37    model=AnthropicModel(model_id="claude-sonnet-4-20250514"),
38    tools=[search_knowledge_base],
39    interventions=[DepartmentAuth(user_dept="engineering")],
40    session_manager=FileSessionManager(session_dir="./sessions"),
41    system_prompt="你是企業知識助手。只根據搜尋結果回答，不要編造。"
42)

 1from strands import Agent
 2from strands.models.anthropic import AnthropicModel
 3
 4model = AnthropicModel(model_id="claude-sonnet-4-20250514")
 5
 6# 文獻搜尋 Agent
 7literature_agent = Agent(
 8    model=model,
 9    name="literature_searcher",
10    description="搜尋學術文獻並提供摘要。輸入研究主題，輸出相關文獻清單與摘要。",
11    system_prompt="你是學術文獻搜尋專家。搜尋相關文獻並提供結構化摘要。",
12    tools=[pubmed_search, arxiv_search]  # 假設已定義
13)
14
15# 資料分析 Agent
16analysis_agent = Agent(
17    model=model,
18    name="data_analyst",
19    description="分析研究數據並產生統計結果。輸入資料集路徑，輸出分析報告。",
20    system_prompt="你是統計分析專家。使用適當的統計方法分析資料。",
21    tools=[python_executor, plot_generator]  # 假設已定義
22)
23
24# 報告撰寫 Agent
25writer_agent = Agent(
26    model=model,
27    name="report_writer",
28    description="將研究成果整理成結構化報告。輸入分析結果，輸出完整報告。",
29    system_prompt="你是科學寫作專家。將研究結果整理成 IMRAD 格式的報告。"
30)
31
32# 指揮 Agent
33research_pipeline = Agent(
34    model=model,
35    system_prompt="""你是研究計畫主持人。根據使用者的研究主題：
36    1. 先用 literature_searcher 搜尋相關文獻
37    2. 用 data_analyst 分析相關數據
38    3. 用 report_writer 撰寫最終報告
39    按順序分派工作，確保每步的輸出能銜接下一步的輸入。""",
40    tools=[literature_agent, analysis_agent, writer_agent]
41)
42
43result = research_pipeline("研究 CRISPR 在腎細胞癌中的應用")

 1from strands import Agent, InterventionHandler
 2from strands.interventions.actions import Proceed, Confirm, Deny
 3from strands.hooks import BeforeToolCallEvent
 4from strands.sandbox import DockerSandbox
 5
 6class DeploymentGuard(InterventionHandler):
 7    """部署護欄：關鍵操作需要人類確認。"""
 8    name = "deploy-guard"
 9    
10    def before_tool_call(self, event: BeforeToolCallEvent, **kwargs):
11        tool_name = event.tool.tool_name
12        
13        # 生產環境部署需要人類確認
14        if tool_name == "deploy_to_production":
15            return Confirm(
16                message="即將部署到生產環境，請確認是否繼續",
17                timeout=300  # 5 分鐘超時
18            )
19        
20        # 刪除操作直接封鎖
21        if tool_name in ["drop_database", "delete_cluster"]:
22            return Deny(reason="破壞性操作已被安全政策禁止")
23        
24        return Proceed()
25
26# 使用 Docker 沙箱隔離部署操作
27deploy_agent = Agent(
28    tools=[run_tests, build_image, deploy_to_staging, deploy_to_production],
29    interventions=[DeploymentGuard()],
30    sandbox=DockerSandbox(container="deploy-sandbox"),
31    system_prompt="你是 DevOps 助手。執行 CI/CD 管線：測試 → 建置 → staging → production。",
32    checkpointing=True  # 啟用檢查點，允許外部系統控制流程
33)

 1from strands import Agent, tool, InterventionHandler, Plugin, Sandbox
 2from strands.models.anthropic import AnthropicModel
 3from strands.models.openai import OpenAIModel
 4from strands.models.ollama import OllamaModel
 5from strands.models.bedrock import BedrockModel
 6from strands.hooks import (
 7    BeforeInvocationEvent, AfterInvocationEvent,
 8    BeforeModelCallEvent, AfterModelCallEvent,
 9    BeforeToolCallEvent, AfterToolCallEvent,
10    MessageAddedEvent, AgentInitializedEvent
11)
12from strands.interventions.actions import Proceed, Deny, Guide, Transform, Confirm
13from strands.session import FileSessionManager, S3SessionManager
14from strands.memory import MemoryManager
15from strands.sandbox import DockerSandbox, PosixShellSandbox
16from strands.plugins import Plugin, hook

 1AgentInitializedEvent          # Agent 建構完成
 2  |
 3BeforeInvocationEvent          # 開始處理請求
 4  |
 5  +-- BeforeModelCallEvent     # 呼叫模型前（每次迴圈）
 6  |     |
 7  |   AfterModelCallEvent      # 模型回應後
 8  |     |
 9  |   BeforeToolCallEvent      # 工具執行前（如需要）
10  |     |
11  |   AfterToolCallEvent       # 工具執行後
12  |     |
13  |   MessageAddedEvent        # 訊息加入對話歷史
14  |     |
15  +-- (迴圈直到 end_turn)
16  |
17AfterInvocationEvent           # 請求處理完成