Strands Agentsで輸入税番検索する

2026年04月26日

今日はプログラミング関連のネタ。

Strands Agentを使って開発した輸入税番検索エージェントを動かしてみた…という話。

うお、Amazon bedrock Strands Agents + GenU AgentCoreで輸入関税検索エージェント、できた！！ pic.twitter.com/4bfDN3d1tz
— qlitre（くりったー） (@kuri_tter) April 25, 2026

輸入税番とは

国内に貨物を輸入する際に税関に申告する商品コードのこと。この商品コードごとに関税率が定められている。

生きている動物、調整食料品などのカテゴリに分かれていて、全部で9000種類以上ある。

ちなみに昨年にhonoを用いて輸入税番検索MCPサーバープログラムを書いた。

それを題材としてCloudflare Workers Tech Talkで発表したことがある。

Cloudflare Workers Tech Talks in Tokyo #6で発表した！

輸入税番に関してはその時の資料に簡単にまとめている。スライド画像を引用する。

https://speakerdeck.com/qlitre/shu-ru-shi-xing-guan-shui-lu-biao-womcpsabahua-sitemita

検索ツールは世の中にぼちぼち出てきているが、まだまだ人力で行っている場合がほとんど。9000種類以上あるので結構大変だ。

これをMCPサーバー化すれば効率化できる…という内容の発表だった。

モチベーションと経緯

発表したMCPサーバーで機能的には実現ができた。しかし、この時に開発したMCPサーバーは自分の個人Cloudfareアカウントにデプロイされている。そのため業務で使うのは厳しいという課題がある。

そこでAWSのリソースで同じようなことをすればセキュアに業務で使えるのでは、ということでやってみたわけだ。

内容

ソースコードはgithubに。

https://github.com/qlitre/aws-tariff-search-agent/

Cloudflareデプロイの場合はそのままWorkersにタリフデータのjsonファイルをアップロードしたが、AgentCoreの場合はそういう領域があるのかわからなかったので、S3にファイルをアップロードした。適当なS3バケットにtariffdata/というプリフィックスでファイルを保存する。

あとはJsonファイルを読み取ってキーワード検索をするAgentプログラムを書く。

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from strands import Agent, tool
from strands.models import BedrockModel
from tariff_service import TariffSearchService
from pathlib import Path
import json
import logging
import boto3

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# タリフ検索サービスのインスタンス（遅延初期化）
_search_service = None


def get_search_service():
    """遅延初期化でTariffSearchServiceのインスタンスを取得"""
    global _search_service
    if _search_service is None:
        _search_service = TariffSearchService()
    return _search_service


@tool
def search_tariff_by_keywords(keywords: str) -> str:
    """Search tariff data by Japanese keywords (comma-separated).

    Use Japanese keywords for best results. Multiple keywords can be separated by commas.

    Args:
        keywords: Comma-separated Japanese keywords to search for

    Returns:
        JSON string containing search results with hitCount per keyword and limited results
    """
    if not keywords or not keywords.strip():
        return json.dumps({
            "error": "keywords is required"
        }, ensure_ascii=False, indent=2)

    try:
        # キーワードをリストに分割
        keyword_list = [k.strip() for k in keywords.split(',') if k.strip()]

        # search_tariff_dataは(results, hit_count)のタプルを返す
        search_service = get_search_service()
        results, hit_count = search_service.search_tariff_data(keyword_list)

        limit = 30
        message = ""
        if len(results) > limit:
            message = f"More than the maximum limit of {limit} items were found. Please refer to hitCount and re-search if necessary."

        return json.dumps({
            "keywords": keywords,
            "found": len(results),
            "message": message,
            "hitCount": hit_count,  # 各キーワードごとのヒット数の辞書
            "results": results[:limit]
        }, ensure_ascii=False, indent=2)
    except Exception as e:
        return json.dumps({
            "error": f"検索エラー: {str(e)}"
        }, ensure_ascii=False, indent=2)


SYSTEM_PROMPT = (Path(__file__).parent / "system_prompt.txt").read_text(encoding="utf-8")

app = FastAPI()


@app.get("/ping")
async def ping():
    """AgentCore required health check"""
    return {"status": "healthy"}


@app.post("/invocations")
async def invocations(request: Request):
    body = await request.body()
    request_data = json.loads(body.decode())

    # promptはStrandsContentBlock[]形式で来る
    prompt = request_data.get("prompt", [])
    model_info = request_data.get("model", {})

    # content blockのリストをStrandsが扱える形式に変換
    if isinstance(prompt, list):
        processed_prompt = ''.join([
            block.get('text', '') for block in prompt
            if isinstance(block, dict)
        ])
    else:
        processed_prompt = str(prompt)

    model_id = model_info.get("modelId", "us.anthropic.claude-sonnet-4-5-20250929-v1:0")
    region = model_info.get("region", "us-west-2")

    bedrock_model = BedrockModel(
        model_id=model_id,
        boto_session=boto3.Session(region_name=region),
    )

    agent = Agent(
        tools=[search_tariff_by_keywords],
        system_prompt=SYSTEM_PROMPT,
        model=bedrock_model,
    )

    async def generate():
        try:
            async for event in agent.stream_async(processed_prompt):
                if "event" in event:
                    yield json.dumps(event, ensure_ascii=False) + "\n"
        except Exception as e:
            logger.error(f"Error: {e}", exc_info=True)
            yield json.dumps({
                "event": {"internalServerException": {"message": str(e)}}
            }, ensure_ascii=False) + "\n"

    return StreamingResponse(generate(), media_type="text/event-stream")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8080, log_level="warning", access_log=False)

システムプロンプト

あなたは日本の関税データを検索する専門エージェントです。

# 利用可能なツール

## search_tariff_by_keywords（関税データ検索）
統計コード・HSコード・税率など構造化された関税データをキーワードで検索します。
- **重要**: ユーザーのメッセージから連想される日本語キーワードを複数カンマ区切りで渡してください。
- 統計コード6桁での検索も可能です。
- 例: 「チューハイの税番を教えて」→ search_tariff_by_keywords(keywords="チューハイ,酒,アルコール,飲料")

# 検索戦略

1. **初回検索**: 商品名・カテゴリ・用途・素材など関連キーワードを3〜5個使用
2. **結果がない場合**: 類似語・上位カテゴリ・別表現で再検索
   - 例: 「スマートフォン」→「携帯電話,電話機,通信機器」
3. **結果が多すぎる場合**（30件以上）: `hitCount`でヒット数の少ないキーワードを確認し、より具体的なキーワードで絞り込む

# 回答形式

<回答形式>
## 検索結果

| 統計コード | HSコード | 品名 | 税率 | 単位 |
|-----------|---------|------|------|------|
| （検索結果を表形式で記載） |

## 解説・補足
（分類の根拠、注意事項、関連法令など）

[その他の有用な情報や提案]
</回答形式>

あとはtoolkitを使えば簡単にデプロイできる。すごい。

agentcore agentcore configure
agentcore deploy --env=S3_BUCKET_NAME=your-bucket-name

※デプロイ後に実行ロールにS3バケットの読取権限をつける必要がある。

{
  "Effect": "Allow",
  "Action": ["s3:GetObject", "s3:ListBucket"],
  "Resource": [
    "arn:aws:s3:::your-bucket-name",
    "arn:aws:s3:::your-bucket-name/*"
  ]
}

フロントエンド

GenUに外部AgentCoreランタイムとして設定して使っている。

https://github.com/aws-samples/generative-ai-use-cases

これも基本的にはデプロイ後のAgentCoreランタイムのarnを設定ファイルに書けばいいだけで簡単。そうすると上の動画のようにチャット画面から呼び出すことができる。

感想など

前々からStrands Agentsは気になっていて初めて実戦的なプログラムをデプロイすることができてよかった。

開発の感覚としてはCloudflareでMCPサーバーを建てた時に近い。AIにやってほしい機能をツールとして持たせるのがコツだ。今回のケースだと税番を特定したい。だから、キーワードを受け取って検索ができるプログラムをツールとして持たせる。そうすると検索キーワードはツールを使うLLMが「よしなに」生成をしてくれる。あとは見つかるまでエージェントが頑張ってくれる…という感じ。いろんな検索キーワードを試して結果を確認して…という面倒な業務を肩代わりしてくれるイメージだろうか。

ちなみにもともとBedrockには「エージェントビルダー」と呼ばれるエージェント作成機能がある。マネコンから簡単にエージェントが作れる、という触れ込みだったと思う。こちらで同様のエージェントを作ったことがあるが、Lambda関数を別途用意する必要があったり、Lambda関数との接続にクセがあったりで意外と落とし穴が多かった印象。

今回AgentCore x Strands Agentsを使ってみた感想として圧倒的にこちらの方がDXがよい。というのもソースコード内でプロンプトから利用ツールまで完結できるので、管理が簡単だ。deployもtoolkitが用意されているので複雑な操作はいらない。インフラのことはあまり意識せずに機能開発に集中することができた。

フロントエンドのGenUからもruntimeのarnを渡せばすぐに使えるようになるので、ツールをどんどん生やしていけそう。例えばAPIが公開されているSaaSと接続させたら面白そうなどと今の時点では考えてる。

そのほか

スライドで１億8972万件輸入申告があったと書いているが、今回のエージェントで解決する税番特定はその件数よりずっと少ないと推測される。なぜなら基本的に前回と同じ商材の場合は過去の結果を使い回すため。
税番特定は特に機械類が難しい。熟練した人物でも特定に１時間程度かかることが多い。
世の中にAI検索系のツールは出てきているが当然市販のものは高い。AgentCoreはサーバーレスなので従量課金のみな点に優位性がある。
詳細な税番は最終的には税関に確認する必要がある

おわりに

エージェント開発面白いです！ではでは。

記事一覧へ