extension-template 完整教學

1# Clone 你新建的 repo
2git clone https://github.com/你的帳號/你的repo名稱
3cd 你的repo名稱

1# 進入 Python 子目錄
2cd python
3
4# 執行互動式 Setup Script
5python setup_template.py

 1Package name (e.g., 'google', 'aws', 'slack'): redis
 2  PyPI package: strands-redis
 3  Module:       strands_redis
 4  Classes:      RedisModel, RedisHooks, etc.
 5
 6Which components do you want to include?
 7  1. Tool - Add capabilities to agents using the @tool decorator
 8  2. Model Provider - Integrate custom LLM APIs
 9  3. Plugin - Extend agent behavior with hooks and tools in a composable package
10  4. Session Manager - Persist conversations across restarts
11  5. Conversation Manager - Control context window and message history
12
13Enter numbers separated by commas (e.g., 1,2): 1,4

1# 安裝為可編輯模式 (editable install)
2pip install -e ".[dev]"
3
4# 或使用 hatch
5hatch env create

1# 進入 TypeScript 子目錄
2cd typescript
3
4# 執行互動式 Setup Script
5npx tsx setup-template.ts
6# 或
7npm run setup

1npm install

1# 執行完整檢查（格式化 → Lint → 型別檢查 → 測試）
2hatch run prepare
3
4# 或個別執行
5hatch run test        # 測試
6hatch run lint        # Linter
7hatch run typecheck   # 型別檢查
8hatch run format      # 格式化

1# 執行完整檢查
2npm run check
3
4# 或個別執行
5npm run test          # 測試
6npm run lint          # ESLint
7npm run type-check    # TypeScript 型別檢查
8npm run format:check  # Prettier 格式檢查

 1"""Tool Implementation."""
 2
 3import logging
 4from typing import Any
 5
 6from strands import tool
 7
 8logger = logging.getLogger(__name__)
 9
10
11@tool
12def template_tool(param1: str) -> dict[str, Any]:
13    """Brief description of what your tool does.
14
15    Args:
16        param1: Description of parameter 1.
17
18    Returns:
19        Dict containing status and response content.
20    """
21    # TODO: Implement your tool logic
22    raise NotImplementedError

 1"""Weather lookup tool for Strands Agents."""
 2
 3import logging
 4from typing import Any
 5import httpx                        # 使用 httpx 做 HTTP 請求
 6
 7from strands import tool
 8
 9logger = logging.getLogger(__name__)
10
11# 1. @tool 裝飾器自動將此函式註冊為 Agent 可呼叫的工具
12#    Strands SDK 會解析 docstring 與 type hints 來產生 tool schema
13@tool
14def get_weather(city: str, units: str = "metric") -> dict[str, Any]:
15    """查詢指定城市的目前天氣狀況。
16
17    Args:
18        city: 要查詢天氣的城市名稱（如 "Taipei"）。
19        units: 溫度單位，"metric" 為攝氏，"imperial" 為華氏。
20
21    Returns:
22        包含溫度、濕度和天氣描述的字典。
23    """
24    # 2. 實作實際的 API 呼叫邏輯
25    api_key = "YOUR_API_KEY"  # 實務上應從環境變數讀取
26    url = f"https://api.openweathermap.org/data/2.5/weather"
27    params = {"q": city, "appid": api_key, "units": units}
28
29    # 3. 發送請求並處理回應
30    response = httpx.get(url, params=params)
31    response.raise_for_status()
32    data = response.json()
33
34    # 4. 回傳結構化結果，Agent 會將此資訊整合到回覆中
35    return {
36        "status": "success",
37        "city": city,
38        "temperature": data["main"]["temp"],
39        "humidity": data["main"]["humidity"],
40        "description": data["weather"][0]["description"],
41    }

1from strands import Agent
2from strands_weather import get_weather  # 你的套件名稱
3
4# 將工具傳入 Agent，Agent 會在需要時自動呼叫
5agent = Agent(tools=[get_weather])
6response = agent("台北現在的天氣如何？")

 1/**
 2 * 自訂 Model Provider 範例 — 整合假想的 "MyLLM" API
 3 */
 4import {
 5  Model,
 6  type BaseModelConfig,
 7  type Message,
 8  type ModelStreamEvent,
 9  type StreamOptions,
10} from '@strands-agents/sdk'
11
12// 1. 定義設定介面，繼承 BaseModelConfig
13export interface MyLLMConfig extends BaseModelConfig {
14  apiKey: string          // API 金鑰
15  endpoint: string        // API 端點 URL
16  temperature?: number    // 生成溫度（可選）
17}
18
19// 2. 繼承 Model 基礎類別，實作必要方法
20export class MyLLMModel extends Model<MyLLMConfig> {
21  private _config: MyLLMConfig
22
23  constructor(config: MyLLMConfig) {
24    super()
25    this._config = { ...config }
26  }
27
28  // 3. 實作 updateConfig：讓 Agent 能在執行期間調整設定
29  override updateConfig(modelConfig: Partial<MyLLMConfig>): void {
30    this._config = { ...this._config, ...modelConfig }
31  }
32
33  // 4. 實作 getConfig：回傳目前設定的副本
34  override getConfig(): MyLLMConfig {
35    return { ...this._config }
36  }
37
38  // 5. 核心方法：實作 stream() 來處理與 LLM 的串流通訊
39  //    必須依序 yield 以下事件：
40  //    ModelMessageStartEvent → ContentBlockStart/Delta/Stop → ModelMessageStopEvent
41  override async *stream(
42    messages: Message[],
43    options?: StreamOptions
44  ): AsyncIterable<ModelStreamEvent> {
45    // 呼叫你的 LLM API
46    const response = await fetch(this._config.endpoint, {
47      method: 'POST',
48      headers: {
49        'Authorization': `Bearer ${this._config.apiKey}`,
50        'Content-Type': 'application/json',
51      },
52      body: JSON.stringify({
53        messages,
54        temperature: this._config.temperature ?? 0.7,
55        tools: options?.toolSpecs,
56      }),
57    })
58
59    // 將 API 回應轉換為 Strands StreamEvent 格式
60    yield { messageStart: { role: 'assistant' } }
61    yield { contentBlockStart: { contentBlock: { type: 'text', text: '' } } }
62
63    // 處理串流回應...
64    const data = await response.json()
65    yield { contentBlockDelta: { delta: { type: 'text', text: data.content } } }
66
67    yield { contentBlockStop: {} }
68    yield { messageStop: { stopReason: 'end_turn' } }
69  }
70}

 1import { Agent } from '@strands-agents/sdk'
 2import { MyLLMModel } from 'strands-myllm'
 3
 4const model = new MyLLMModel({
 5  apiKey: process.env.MYLLM_API_KEY!,
 6  endpoint: 'https://api.myllm.com/v1/chat',
 7  temperature: 0.5,
 8})
 9
10const agent = new Agent({ model })
11const response = await agent.run('幫我寫一首詩')

 1"""Logging Plugin — 記錄所有工具呼叫的執行日誌。"""
 2
 3import logging
 4import time
 5from typing import TYPE_CHECKING
 6
 7from strands import tool
 8from strands.hooks import BeforeToolCallEvent, AfterToolCallEvent
 9from strands.plugins import Plugin, hook
10
11if TYPE_CHECKING:
12    from strands.agent.agent import Agent
13
14logger = logging.getLogger(__name__)
15
16
17class LoggingPlugin(Plugin):
18    """記錄 Agent 工具呼叫的 Plugin。
19
20    自動追蹤每次工具呼叫的：
21    - 工具名稱與輸入參數
22    - 執行耗時
23    - 成功/失敗狀態
24    """
25
26    # 1. name 屬性用於辨識 Plugin
27    name = "logging-plugin"
28
29    def __init__(self, log_level: int = logging.INFO) -> None:
30        super().__init__()
31        self._log_level = log_level
32        self._call_times: dict[str, float] = {}  # 追蹤呼叫開始時間
33
34    # 2. init_agent 在 Plugin 被 Agent 載入時執行
35    #    裝飾了 @hook 和 @tool 的方法會自動被註冊，
36    #    這裡可以做額外的初始化邏輯
37    def init_agent(self, agent: "Agent") -> None:
38        logger.log(self._log_level, "LoggingPlugin initialized for agent")
39
40    # 3. @hook 裝飾器：在工具呼叫前觸發
41    @hook  # type: ignore[call-overload]
42    def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
43        tool_name = event.tool_use.get("name", "unknown")
44        tool_input = event.tool_use.get("input", {})
45
46        # 記錄呼叫開始時間
47        call_id = event.tool_use.get("toolUseId", "")
48        self._call_times[call_id] = time.time()
49
50        logger.log(
51            self._log_level,
52            f"[TOOL CALL] {tool_name} | Input: {tool_input}"
53        )
54
55    # 4. @tool 裝飾器：Plugin 也可以攜帶自己的工具
56    #    注意：Plugin 的 tool 方法第一個參數是 self
57    @tool
58    def get_call_stats(self) -> str:
59        """取得目前的工具呼叫統計資訊。"""
60        total = len(self._call_times)
61        return f"Total tool calls tracked: {total}"

1from strands import Agent
2from strands_logging import LoggingPlugin
3
4# Plugin 透過 plugins 參數傳入，可以傳入多個
5plugin = LoggingPlugin(log_level=logging.DEBUG)
6agent = Agent(plugins=[plugin])
7
8# Agent 執行時，Plugin 的 hook 會自動在對應的時間點觸發
9response = agent("請查詢台北的天氣，然後顯示呼叫統計")

 1/**
 2 * Redis-based SnapshotStorage — 使用 Redis 持久化 Agent 對話
 3 */
 4import type {
 5  Snapshot,
 6  SnapshotLocation,
 7  SnapshotManifest,
 8  SnapshotStorage
 9} from '@strands-agents/sdk'
10import { createClient, type RedisClientType } from 'redis'
11
12export interface RedisStorageConfig {
13  connectionString: string  // Redis 連線字串
14  keyPrefix?: string        // Key 前綴（預設 "strands:"）
15}
16
17export class RedisSnapshotStorage implements SnapshotStorage {
18  private client: RedisClientType
19  private prefix: string
20
21  constructor(private readonly config: RedisStorageConfig) {
22    // 1. 建立 Redis 連線
23    this.client = createClient({ url: config.connectionString })
24    this.prefix = config.keyPrefix ?? 'strands:'
25  }
26
27  // 2. 儲存快照：將 Snapshot 序列化為 JSON 存入 Redis
28  async saveSnapshot(params: {
29    location: SnapshotLocation
30    snapshotId: string
31    isLatest: boolean
32    snapshot: Snapshot
33  }): Promise<void> {
34    const key = `${this.prefix}${params.location.sessionId}:${params.snapshotId}`
35    await this.client.set(key, JSON.stringify(params.snapshot))
36
37    // 如果是最新的，額外存一份 "latest" 指標
38    if (params.isLatest) {
39      const latestKey = `${this.prefix}${params.location.sessionId}:latest`
40      await this.client.set(latestKey, params.snapshotId)
41    }
42  }
43
44  // 3. 載入快照：從 Redis 讀取並反序列化
45  async loadSnapshot(params: {
46    location: SnapshotLocation
47    snapshotId?: string
48  }): Promise<Snapshot | null> {
49    let snapshotId = params.snapshotId
50    if (!snapshotId) {
51      // 沒指定 ID 就載入最新的
52      const latestKey = `${this.prefix}${params.location.sessionId}:latest`
53      snapshotId = await this.client.get(latestKey) ?? undefined
54    }
55    if (!snapshotId) return null
56
57    const key = `${this.prefix}${params.location.sessionId}:${snapshotId}`
58    const data = await this.client.get(key)
59    return data ? JSON.parse(data) : null
60  }
61
62  // 4. 其他必要方法...
63  async listSnapshotIds(params: {
64    location: SnapshotLocation
65    limit?: number
66  }): Promise<string[]> {
67    const pattern = `${this.prefix}${params.location.sessionId}:*`
68    const keys = await this.client.keys(pattern)
69    return keys
70      .map(k => k.replace(`${this.prefix}${params.location.sessionId}:`, ''))
71      .filter(k => k !== 'latest' && k !== 'manifest')
72      .slice(0, params.limit)
73  }
74
75  async deleteSession(params: { sessionId: string }): Promise<void> {
76    const pattern = `${this.prefix}${params.sessionId}:*`
77    const keys = await this.client.keys(pattern)
78    if (keys.length > 0) await this.client.del(keys)
79  }
80
81  async loadManifest(params: {
82    location: SnapshotLocation
83  }): Promise<SnapshotManifest> {
84    const key = `${this.prefix}${params.location.sessionId}:manifest`
85    const data = await this.client.get(key)
86    return data ? JSON.parse(data) : { snapshotIds: [] }
87  }
88
89  async saveManifest(params: {
90    location: SnapshotLocation
91    manifest: SnapshotManifest
92  }): Promise<void> {
93    const key = `${this.prefix}${params.location.sessionId}:manifest`
94    await this.client.set(key, JSON.stringify(params.manifest))
95  }
96}

 1import { Agent, SessionManager } from '@strands-agents/sdk'
 2import { RedisSnapshotStorage } from 'strands-redis'
 3
 4// SessionManager 是 SDK 內建的，你只需提供 storage backend
 5const session = new SessionManager({
 6  sessionId: 'user-123-session',
 7  storage: {
 8    snapshot: new RedisSnapshotStorage({
 9      connectionString: 'redis://localhost:6379',
10      keyPrefix: 'myapp:',
11    }),
12  },
13})
14
15const agent = new Agent({ sessionManager: session })

1[tool.hatch.version]
2source = "vcs"
3
4[tool.hatch.version.raw-options]
5root = ".."                                    # Monorepo 模式：Git root 在上層
6tag_regex = "^python-v(?P<version>.+)$"        # 匹配 python-v0.1.0 格式
7relative_to = "pyproject.toml"                 # 參考點

 1import { tool } from '@strands-agents/sdk'
 2import { z } from 'zod'
 3
 4// Zod schema 定義了工具的輸入參數
 5const inputSchema = z.object({
 6  param1: z.string().describe('Description of parameter 1.'),
 7})
 8
 9// tool() factory 接受 Zod schema，自動產生 JSON Schema 並做執行期驗證
10export const templateTool = tool({
11  name: 'template_tool',
12  description: 'Brief description of what your tool does.',
13  inputSchema,
14  callback: (input: z.infer<typeof inputSchema>) => {
15    // input 已經過 Zod 驗證，型別安全
16    return { result: input.param1.toUpperCase() }
17  },
18})

 1class TemplatePlugin(Plugin):
 2    # @hook 裝飾的方法會被 Plugin Registry 自動掃描並註冊
 3    @hook
 4    def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
 5        pass
 6
 7    # @tool 裝飾的方法同樣會被自動發現
 8    @tool
 9    def my_plugin_tool(self, param1: str) -> str:
10        pass

 1class TemplatePlugin implements Plugin {
 2    // 在 initAgent 中顯式註冊 hook
 3    initAgent(agent: LocalAgent): void {
 4        agent.addHook(BeforeToolCallEvent, (event) => { ... })
 5    }
 6    // 在 getTools 中顯式回傳工具
 7    getTools(): Tool[] {
 8        return [templatePluginTool]
 9    }
10}

1sdk-python / sdk-typescript (核心 SDK)
2    ↑ 相依
3extension-template (起始模板)
4    ↓ 產出
5社群套件 (如 strands-redis, strands-slack)
6    ↓ 被使用
7Agent 應用程式 (透過 agent-builder / samples / multiagent)

1from strands import Agent
2from strands_internal_llm import InternalModel
3
4model = InternalModel(
5    endpoint="https://llm.internal.company.com/v1",
6    api_key="internal-token",
7    model_id="meta-llama/Llama-3.1-70B",
8)
9agent = Agent(model=model)

1from strands import Agent
2from strands_slack import SlackPlugin, send_slack_message, read_slack_channel
3
4agent = Agent(
5    tools=[send_slack_message, read_slack_channel],
6    plugins=[SlackPlugin(workspace="my-company")]
7)
8agent("幫我在 #general 頻道發送一則公告")

 1import { Agent, SessionManager } from '@strands-agents/sdk'
 2import { DynamoSnapshotStorage } from 'strands-dynamodb'
 3
 4const session = new SessionManager({
 5  sessionId: req.headers['x-session-id'],
 6  storage: {
 7    snapshot: new DynamoSnapshotStorage({
 8      tableName: 'agent-sessions',
 9      region: 'ap-northeast-1',
10    }),
11  },
12})
13
14const agent = new Agent({ sessionManager: session })
15// 即使 Lambda cold start，也能從 DynamoDB 恢復完整對話歷史