はじめに
Claude Code上で、業務の一部をAIエージェントによって自動化する取り組みが広がっています。
こうしたAIエージェントは、自然言語で記述するAgent Skillsを使用することで、非エンジニアでも構築しやすくなります。
AIの民主化という観点では、とても良いことだと思います。
ただ、業務データを利用する例は、あまり見かけない気がします。
そこで、業務データを利用するAIエージェントを、スキルを組み合わせて作りました。
今回やったこと
架空の健康食品販売会社を想定しました。
顧客からのWebフォームでの問い合わせ対応業務の一次処理部分を担うAIエージェントを作りました。
Claude Code上で動きます。
このAIエージェントは、3つのスキルで構成されます。
そして、様々な業務データを選択・収集し、読み込み、次々と答えを出していきます。
「Webフォームでの問い合わせ」の内容は、この会社の顧客からのクレーム・質問・依頼です。
ここで言う「問い合わせ対応業務の一次処理部分」とは、
①まだ対応未着手の問い合わせを1つずつ見て、どう対応すべきか、という対応案を策定
②そのうち、緊急を要する問い合わせは、直ちに社内に通知
を指します。
上記2つを、AIエージェントが定期的に自動で行います。
想定している読者
非エンジニアの方でも十分理解可能です。
- 業務データを利用するAIエージェントを業務フローに組み込む簡単な例が知りたい人
- Agent SkillsやMCPについて、概念レベルで知っている人
- Claude Codeを普段から使用している人
本稿の立ち位置
本稿は
「AIエージェントが既存の業務データをどう使って、何をもたらすか」
「そのために、どんな作りにしてあるか」
を見せて、今後のヒントにしてもらうことを目的にしています。
ハンズオン記事ではありませんので、具体的な手の動かし方や細かいHow to、技術論はほとんどありません。
※このAIエージェントでは、実行の手順自体はこちらが決めています。
決めた通りに実行すれば目的が達成でき、わざわざ手順までAIに考えさせる必要がないためです。
ただし、その手順の一つ一つの中で、AIが必要と思われるデータを集めて結論を下す、ということを繰り返します。
AIエージェントの概要
目指すもの
[問い合わせ対応担当者の要望]
Webフォームでの問い合わせに対応する一次処理とは、どういうものでしょうか。
ざっと挙げていくと、こんな感じでしょう。
-
どんな類の問い合わせか、判別する。
クレームか、質問か、依頼か。 -
問い合わせしてきたこの顧客の”重み”を判断する。
当社との関係性はどうか。この顧客は手厚く扱うべきか。注意して対応すべきか。 -
機械的に回答できるか、人間が対応すべきか、判断する。
FAQや規約等を読めば機械的に回答できる簡単な質問か、それとも人間が対応する必要があるものか。
・機械的に回答可能 → 回答文を作成する。
・人間が対応 → 対応方針を立てる。 -
過去の類似事案を調べる。
過去、類似の事案にどう対応したか、参考にするため。 -
緊急で対応すべきか、判断する。
緊急なら、しかるべき人員に急いで通知する。
結局、どんな問い合わせでも、一次処理では、やることは大体同じです。
であれば、
随時、AIエージェントにこれらを全部やらせて、省力化を図ろう
というのが、目指すところです。
[目指すもの ～ 人力対応からAIエージェントへ]
AIエージェント担当部分＆仕事の内容
そのAIエージェントが担当するのは、下図の赤枠点線の部分です。
定期実行されます。
- 対応未着手の各問い合わせに対して、様々な観点から、対応案を策定
策定された対応案は、その問い合わせデータに記録される - 緊急性有りと判断された問い合わせは、一括して、社内にSlackで通知
[AIエージェント担当部分＆仕事の内容]
対顧客なので、責任は人間が持ちます。
AIは、「案」を出すだけです。
最終決定は、人間が行います。
従って、メール自動送信で回答する場合でも、AI作成の回答文案は、必ず人間が送信前にチェックする、という想定です。
（回答メール自動送信は、AIエージェントの仕事ではありません）
 業務要件
|
 機能要件 |
|---|---|
| 対応未着手の問い合わせ１件ごとに、対応案が欲しい。 |
AIは、対応未着手の問い合わせ１件ごとに、以下を提案・提示する。 ・問い合わせの種類（質問/依頼/クレーム） ・自動送信メールでの回答で済ませるか or 人間が対応すべきか ・問い合わせした顧客の重要度 ・過去の類似事案の対応の実例 ・緊急性の有無 ⇒ ・上記から導出される対応方針 |
| 緊急性の有るものは、遅滞なく社内に展開したい。 | AIは、緊急性有りと判断されたものを、その判断の直後に、社内のSlack専用チャンネルに一括投稿する。 |
| 未対応の問い合わせが滞留しないようにしたい。 |
AIは、ある程度まとまった件数の対応未着手の問い合わせを一括処理する。 それを、定期的に実行する。 |
| 最終決定は人間が行いたい。 |
AIのアウトプットは、「案」にとどめる。 対応そのものや、回答メールの自動送信などは、行わない。 |
全体の論理構成と役割分担
AIエージェントは、以下のように、頻繁にDBにアクセスします。
- まず、対応未着手の問い合わせデータをDBから取得
- その問い合わせへの対応を考えるために、DB内の各情報ソースにアクセス
- 最後に、考えた対応案を、当該問い合わせデータに記録
そのため、AIエージェントとDBの間に、MCPサーバーを嚙ませます。
＜ AIエージェント ＞
問い合わせ対応を策定する頭脳部分です。
加えて、緊急性のある問い合わせを社内に通知することも担います。
中身は、各々やるべき仕事を持つ「サブエージェント」2つの直列実行パイプラインです。
メインエージェント（我々の要求を直接受け付けるAI）は、
サブエージェント1：問い合わせの対応案の策定
↓
サブエージェント2：緊急性の有る問い合わせの社内通知
と、直列実行します。
メインエージェントは、1個の専用のスキルの指示に従います。
その専用スキル内で、明示的にサブエージェント1と2を直列実行する、という指示がなされています。
「サブエージェント」とは、メインエージェントが呼び出す別の下請けAIです。
呼び出されると、独立した作業領域で仕事をこなし、結果だけをメインエージェントに返します。
各サブエージェントにも、1個だけ専用スキルが設定されています。
その専用スキルに、そのサブエージェントがすべき仕事が書かれています。
各サブエージェントは、その専用スキルに書いてある通りに仕事をします。
| 実行順序 | エージェント | やるべき仕事 |
|---|---|---|
| - | メインエージェント | 専用スキル「スキル0」の指示： サブエージェント1→サブエージェント2と、直列実行する。 |
| 1 | サブエージェント1 | 専用スキル「スキル1」の指示： 未着手の問い合わせに対して、対応案を策定する。 |
| 2 | サブエージェント2 | 専用スキル「スキル2」の指示： 緊急性有りと判断された問い合わせ全件を、社内に通知する。 |
＜ MCPサーバー群 ＞
各MCPサーバーの機能と用途は、以下の通りです。
| MCPサーバー | 機能 | 用途 |
|---|---|---|
| MCPサーバー1 | 販売データの統計を返す。 |
[この客は手厚く対応すべき上客か、の判断に使用] サブエージェント1が対応案を策定する際、 「この顧客は今までどれほど買ってくれた客か、手厚く対応すべき上客か」 を判断するために使用される。 |
| MCPサーバー2 | 問い合わせ関連データの読み書き。 |
・[問い合わせデータの準備と更新に使用] サブエージェント1は、対応案を策定する際、対応未着手の問い合わせデータをごそっと読み込む。 そして対応案策定後、当該問い合わせデータに、対応案やその他の情報を書き込む。 このように、問い合わせデータの準備と更新に使用される。 ・[この客はケアすべき客か、の判断に使用] サブエージェント1が対応案を策定する際、 「この顧客との間に、今までどのようなやり取りがあり、どういう結果になったのか、ケアすべき（注意して扱うべき）客か」 を、問い合わせ対応データから判断するために使用される。 ・[参考にすべき先行対応事例の収集に使用] サブエージェント1が対応案を策定する際、 類似事案の対応事例を調べて、対応案の参考にする。 そのために問い合わせ対応データを読むのに使用される。 |
| MCPサーバー3 | 緊急性の有る問い合わせをSlackチャンネルに一括送信。 |
[社内Slackチャンネルへの送信ツールとして使用] サブエージェント2は、緊急性の有る問い合わせ全件を社内のSlackチャンネルに一括送信して、迅速な対応を促す。そのためのSlack送信ツールとして使用される。 |
＜ DB ＞
問い合わせ関連データ（注文データ、問い合わせデータ、問い合わせ対応データ）と、
販売データ
があります。
注文データ1件には、
顧客、注文した商品やその価格、配送に関する情報（いつ発送する（した）か、いつ到着する（した）か、など）、
などがあります。
問い合わせ1件は、注文1件に対してなされます。
つまり、問い合わせデータ1件は、必ず注文データ1件に紐づきます。
問い合わせの内容は、クレーム／質問／依頼です。
・クレームの例）注文したXXが、到着予定日を過ぎても届かない！
・クレームの例）今日届いたXXのパッケージが破損している！
・質問の例）おととい注文したXXは、キャンセルできますか？
・質問の例）昨日注文したXXは、いつ届く予定ですか？
問い合わせデータ1件には、
問い合わせした顧客、紐づく注文の注文番号、問い合わせ本文、対応着手済みフラグ、
問い合わせ本文要約、対応ルート案（人間が対応or自動送信メールで回答、のどちらがいいのか）、緊急性の有無フラグ、対応のヒント、自動送信メール回答文案、
など多数の情報があります。
この色は、サブエージェント1が、「対応案策定」として決定するものです。
問い合わせ対応データ1件は、
問い合わせに対する実対応が完了した時点で生成されます（本稿のAIエージェント範囲外）。
つまり、問い合わせ対応データ1件は、必ず問い合わせデータ1件に紐づきます。
問い合わせ対応データ1件には、
紐づく問い合わせの問い合わせ番号、実際の対応内容、
などがあります。
販売データ１件には、
販売先の顧客に、いつ、何をどのくらい、いくらで売ったか、
などがあります。
AIエージェントの構成と各部のはたらき
【前提】
- DBとMCPサーバー群は別の場所に存在し、MCPサーバー群は常時稼働中である、とします。
[全体の論理構成と役割分担]
[AIエージェント内のエージェントとスキルの構成]
【参考】AIエージェントPJのディレクトリ構成（クリックで開閉）
(PJルート)/
│
├── .claude/
│ │
│ ├── settings.json
│ │
│ ├── agents/
│ │ │
│ │ ├── sub-res-plan.md (サブエージェント1のmdファイル)
│ │ │
│ │ └── sub-inquiry-urg-ntf.md (サブエージェント2のmdファイル)
│ │
│ └── skills/
│ │
│ ├── skill-main-inquiries-res/
│ │ │
│ │ └──SKILL.md (メインエージェントが実行する専用スキル「スキル0」)
│ │
│ ├── skill-res-plan/
│ │ │
│ │ ├──SKILL.md (サブエージェント1が実行する専用スキル「スキル1」)
│ │ │
│ │ └── references/
│ │ │
│ │ └── shopping-guide.md
│ │
│ └── skill-inquiry-urg-ntf/
│ │
│ └──SKILL.md (サブエージェント2が実行する専用スキル「スキル2」)
│
└── .mcp.json
メインエージェント
我々の要求を直接受け付けるAIです。
2つのサブエージェントを、直列実行します。
メインエージェントが実行すること
メインエージェントは、「スキル0」（正式名称「skill-main-inquiries-res」）を実行します。
スキル0の指示は、
サブエージェント1：問い合わせ対応案の策定
サブエージェント2：緊急性の有る問い合わせを社内に通知
を、直列実行することだけです。
それ以外には、基本的に何もしません。
※AIエージェントを定期実行する際、このスキル0（「skill-main-inquiries-res」）を名指しで実行します。自然言語による発火ではありません。
スキル0の中身
ステージ1：サブエージェント1（問い合わせ対応案の策定）の起動
ステージ2：サブエージェント2（緊急性の有る問い合わせを社内に通知）の起動
を、直列実行します。
スキル0「skill-main-inquiries-res」のSKILL.md
---
name: skill-main-inquiries-res
description: 健康食品販売会社への問い合わせの対応案策定と、緊急を要する問い合わせの社内通知を行う。
健康食品販売会社の問い合わせ対応案策定パイプラインを直列実行するメインスキル。
このスキルでは、サブエージェントの起動のみ行う。
disable-model-invocation: true
---
## 役割
このスキルはサブエージェントの起動のみを担う。
## 実行手順
以下を必ず直列で実行すること。前のステージの完了を確認してから次を起動する。
いずれかのステージでサブエージェントからエラーなど異常の報告を受けた場合、それを表示して、本スキルの処理を中断する。
### ステージ1
`sub-res-plan` サブエージェントを起動する。
### ステージ2
`sub-inquiry-urg-ntf` サブエージェントを起動する。
## 完了報告
全ステージ完了後、各エージェントから受け取った処理結果をまとめて報告する。
skill-main-inquiries-res：スキル0の名前です。
sub-res-plan：サブエージェント1（問い合わせ対応案の策定）の名前です。
sub-inquiry-urg-ntf：サブエージェント2（緊急性の有る問い合わせを社内に通知）の名前です。
サブエージェント1
ここは、このAIエージェントの「核」となる部分です。
問い合わせの対応案を策定します。
サブエージェント1が実行すること
「スキル1」（正式名称「skill-res-plan」）の指示に従って、以下を実行します。
- 対応未着手の問い合わせデータを最大15件取得する。
- 問い合わせデータ1件ごとに、いろんな情報を参照しつつ、対応案を策定し、その対応案を当該問い合わせデータに書き込む。→対応着手済みとなる。
サブエージェント1が決めること
問い合わせデータ1件ごとの、対応案の「策定」の中身は、以下のものです。
まず最初に、サブエージェント1は、
「問い合わせ本文の要約」と、
「対応ルート」（人間が対応 or 自動送信メールで回答）
の決定を行います。
| AI決定項目 | 内容・意味 |
|---|---|
| 問い合わせ本文の要約 | 問い合わせ本文の要約。 |
| 対応ルート |
「人間が対応」 or 「自動送信メールで回答」 のどちらのルートで対応するか。 以下の基準で決定。 １．回答が自明であり、誰が回答しても、本質的に同じである ・ショッピングガイド（ルールや規約）に、この問い合わせに対する答えがある ・問い合わせの対象である注文データに、問い合わせに対する答えがある ⇒ 「自動送信メールで回答」と決定。 ２．その他全て ⇒ 「人間が対応」と決定。 |
以降は、対応ルートによって、サブエージェント1が決める項目は異なります。
| AI決定項目 | 内容・意味 | 対応ルート= 「人間が対応」 |
対応ルート= 「自動送信メールで回答」 |
|---|---|---|---|
| 対応のヒント (核となる項目) |
対応方針。 加えて、その根拠として、以下の情報がまとめられている。 ・この顧客に対する販売実績 →この客は手厚く対応すべき客か？ ・この顧客との過去のやり取り →この客はケアすべき客か？ ・類似事案の対応履歴 →参考にすべき先行対応事例 |
決定される | 決定されない (null) |
| 緊急性の有無 | 緊急で対応しなければいけない問い合わせか。 以下のいずれかに該当した場合、緊急性有り、と判断される。 ・顧客が怒っている、または強い不満を示している ・商品の破損・欠陥を訴えている ・誤配送・納期遅延を訴えている ・安全に関わる問題である ・過去、この顧客に多大な迷惑をかけている ・この顧客はかなりの上客である |
決定される | 決定されない (null) |
| 自動送信メールの回答の本文案 | 自動送信メールで回答する場合、その回答の本文案。 | 決定されない (null) |
決定される |
[サブエージェント1が実行するスキル1の流れ図]
※下記のスキル1は、全て自然言語のみで記述されている。
サブエージェント1が参照する情報とその用途
対応案策定に際し、サブエージェント1は、以下の情報を収集します。
| 情報 | 用途 | 介在するMCPサーバー |
|---|---|---|
| 販売データ | この客は手厚く対応すべき上客か、の判定に使用。 | MCPサーバー1 販売統計収集 |
| 問い合わせ対応データ （この顧客の問い合わせ対応履歴） |
この客はケアすべき客か、の判定に使用。 | MCPサーバー2 問い合わせ関連データ管理 |
| 問い合わせ対応データ （類似事案の対応履歴） |
参考にすべき先行対応事例として使用。 | MCPサーバー2 問い合わせ関連データ管理 |
| ショッピングガイド （ルールや規約） |
対応ルートの決定と、自動返信メールの回答の本文案作成に使用。 | （なし） |
[サブエージェント1が参照する情報とその用途]
サブエージェント1の中身
スキル1（「skill-res-plan」）を実行することが書いてあります。
それ以外、大したことは書いていません。
サブエージェント1のmdファイル sub-res-plan.md
---
name: sub-res-plan
description: 各問い合わせへの対応案を策定する。問い合わせ対応案策定パイプラインの第1ステージとして使用する。
skills:
- skill-res-plan
---
`skill-res-plan` の手順に従って処理を実行する。
途中でユーザーへの質問・確認を挟まず、自律的に実行する。
処理中にエラーや異常が発生した場合は、その時点で処理を中止し、エラー内容を出力する。
正常に完了した場合は、実行結果を出力する。
sub-res-planは、サブエージェント1の名前です。
skill-res-planは、スキル1の名前です。
スキル1の中身
問い合わせ対応案の策定処理の詳細が書かれています。
問い合わせ対応案の策定を実際に行う中心部です。
スキル1「skill-res-plan」のSKILL.md
---
name: skill-res-plan
description: 未着手の問い合わせに対し、対応案を策定する。その対応案を当該問い合わせデータに反映する。
---
# 役割
- 未着手の問い合わせに対し、対応案を策定する。
- その対応策を当該問い合わせデータに反映する。
## 手順
1. ステップ1：対象となる問い合わせデータの抽出
2. ステップ2：1件ずつ対応案を策定（2-1 から 2-7 の工程を順に行う）
3. ステップ3：結果報告
個々のステップの詳細は下記の通り。
## ステップ1：対象となる問い合わせデータの抽出
未着手の問い合わせデータを取得する。問い合わせ日時が古い順（問い合わせ日が同じ問い合わせの場合、問い合わせ番号が若い順）に、最大15件とする。
取得件数が0件の場合、「問い合わせ対応案の策定処理が完了しました。未着手の問い合わせデータはありませんでした。」と出力し、本スキルの処理を終了する。
## ステップ2：1件ずつ対応案を策定
ステップ1で取得した問い合わせデータに対し、1件ずつ順に 2-1 から 2-7 の工程を行う。
1件について全工程が完了してから、次の1件に進む。
### 2-1. 問い合わせ本文の要約の作成
問い合わせ本文を簡潔に要約する。情報を欠落させずにコンパクトにまとめる。
問い合わせ者の訴えが読んですぐにわかるような文体・文章構造にする。
### 2-2. 対応ルートの決定
対応ルートを以下のいずれかに決定する。
- 「人間が対応」
- 「自動送信メールで回答」
#### 判断基準
以下に該当する場合、「自動送信メールで回答」とする。
- 回答が自明であり、誰が回答しても、本質的に同じである
- ショッピングガイド（`references/shopping-guide.md`）に、この問い合わせに対する答えがある
- 問い合わせの対象である注文データに、問い合わせに対する答えがある
上記に該当しない場合、「人間が対応」とする。
### 2-3. 情報の収集（対応ルートが「人間が対応」の場合のみ）
※対応ルートが「自動送信メールで回答」の場合は、この工程を飛ばす。情報の収集は行わない。
対応案を策定するために、以下の観点での情報を収集する。
どの情報をどの程度収集するかは、問い合わせの内容に応じて判断する。
※必ず、最新の情報を収集する。
- 過去の販売実績から見て、他の顧客より手厚く対応すべき上客か。
- 過去の問い合わせを通じた、この顧客と当社の関係性。注意して対応すべき顧客か。
- 参考にすべき先行対応事例。
- ショッピングガイド（`references/shopping-guide.md`）に、この問い合わせに対する答えやヒントがないか。
### 2-4. 対応のヒントの策定（対応ルートが「人間が対応」の場合のみ）
※対応ルートが「自動送信メールで回答」の場合は、この工程を飛ばす。対応のヒントは無しとする。
後に人間がこの問い合わせに対応する際、有益となるであろう情報をまとめる。
#### 対応のヒントの策定方針
2-3 で収集した情報のうち、対応に関係するものを取捨選択し、社員が対応時に参照すべき観点も含めて記述する。
読む人間が読みやすいように、以下のように構造化する。各部分は簡潔な文章にし、適切な見出しをつける。
- どういう問い合わせか、どういう顧客か、参考にすべき先行対応事例はどのようなものか、対応方針
- 最後に、対応方針を明示する。
### 2-5. 緊急性の有無の決定（対応ルートが「人間が対応」の場合のみ）
※対応ルートが「自動送信メールで回答」の場合は、この工程は飛ばす。緊急性の有無は、無し（判定の要なし）とする。
以下のいずれかに該当する場合、「緊急性有り」とする。該当しない場合、「緊急性なし」とする。
- 顧客が怒っている、または強い不満を示している
- 商品の破損・欠陥を訴えている
- 誤配送・納期遅延を訴えている
- 安全に関わる問題である
- 過去、この顧客に多大な迷惑をかけている
- この顧客はかなりの上客である
### 2-6. 自動送信メールの回答の本文案の作成（対応ルートが「自動送信メールで回答」の場合のみ）
※対応ルートが「人間が対応」の場合は、この工程は飛ばす。自動送信メールの回答の本文案は、無しとする。
顧客へ送信するメールの回答文の本文部分の案を作成する。宛先・送信元・署名・あいさつなど、文面の外枠にあたるものは含めない。
根拠がない回答はしないこと。
#### 回答文の作成方針
- 問い合わせに対する直接的な回答を、丁寧かつ簡潔に書く。
- 顧客にとって有益な情報は、直接的な回答にならなくても、積極的に回答に加える。
### 2-7. 問い合わせデータの更新
ここまでで、下記の項目が決定した。
- 問い合わせ本文の要約
- 対応ルート
- 対応のヒント（対応ルートが「人間が対応」の場合のみ）
- 緊急性の有無（対応ルートが「人間が対応」の場合のみ）
- 自動送信メールの回答の本文案（対応ルートが「自動送信メールで回答」の場合のみ）
決定した上記の項目で、当該問い合わせデータ1件を更新する。同時に「着手済」にする。
## ステップ3：結果報告
以下を出力する。
- 「問い合わせ対応案の策定処理が完了しました。」という文言
- 対応ルート「人間が対応」の問い合わせ番号の羅列。そのうち「緊急性有り」の問い合わせ番号の羅列。
- 対応ルート「自動送信メールで回答」の問い合わせ番号の羅列
※スキル1のSKILL.md内で、MCPサーバーへの言及がないことについて
このスキル1では、DBアクセスを要求する処理を頻繁に行います（未着手の問い合わせデータ取得、顧客の販売実績収集、問い合わせデータ更新など）。
にもかかわらず、SKILL.mdでは、DBアクセスを担当するMCPサーバー1と2に関することや、MCPツール名などは、一切書かれていません。
Claudeは、PJ起動時に既にMCPサーバーの情報を取得済みなので、自然言語での指示だけで適切なMCPサーバーとそのツールを自力で選択できます。
なので、書く必要がないのです。
例を挙げます。
スキルのSKILL.md内の
未着手の問い合わせデータを取得する。
を実際に行うのは、
MCPサーバー2inquiry_data_mngの、
ツールget_inquiries
です。
このMCPサーバーとこのツールの機能説明文は、Claude Codeに事前に登録されています。
Claude Codeに事前に登録されている、ツール「get_inquiries」の機能説明文
健康食品の販売システムにおいて、問い合わせデータを検索して返す。
1レコードが1件の問い合わせに相当する。
（以下省略）
Claudeは、AIエージェント実行時には既にこの登録内容を知っているので、
「未着手の問い合わせデータを取得する。」
という自然言語での指示だけで、このツールを選択し、実行することができます。
これは、このスキルの処理が比較的シンプルであり、MCPサーバーの内部構成もまたシンプルだから成り立っているものです。
複雑な処理の場合や、MCPサーバーの役割分担が曖昧になる場合（複数の似た役割のMCPサーバー群、似た役割のツール群など）は、SKILL.md内での明示的な言及が必要になるでしょう。
（サブエージェント1が参照する情報）ショッピングガイドの中身
サブエージェント1は、対応案策定に際し、ショッピングガイドも参照します。
規約やルールが記載されています。
といっても、わずか数行のmdファイルです。
ショッピングガイドのmdファイル shopping-guide.md
# ショッピングガイド
## キャンセル
- 発送されるまでは無料でキャンセル可能です。発送後はキャンセルできません。
- キャンセル料は発生しません。
## 注文の変更
- 商品の変更、数量の変更は、キャンセル扱いとなります。「キャンセル」をご参照ください。
- いったんキャンセルをした後、注文をし直してください。
# お支払い
- 前払いのみとなります。決済完了後、ご注文成立となります。
- お支払方法は、クレジットカードのみとなります。
サブエージェント2
サブエージェント1の実行が完了後に、サブエージェント2の実行がなされます。
サブエージェント2は、緊急性が有る問い合わせをまとめて一括して社内に通知します。
サブエージェント2が実行すること
「スキル2」（正式名称「skill-inquiry-urg-ntf」）の指示に従って、以下を実行します。
- サブエージェント1により緊急性有りと判断された問い合わせの全てを、一括してSlackの専用チャンネルに送信する。
実際にそれを行うのは、MCPサーバー3である。
サブエージェント2の中身
スキル2（「skill-inquiry-urg-ntf」）を実行することが書いてあります。
それ以外、大したことは書いていません。
サブエージェント2のmdファイル sub-inquiry-urg-ntf.md
---
name: sub-inquiry-urg-ntf
description: 緊急性ありの問い合わせを社内通知する。問い合わせ対応案策定パイプラインの第2ステージとして使用する。
skills:
- skill-inquiry-urg-ntf
---
`skill-inquiry-urg-ntf` の手順に従って処理を実行する。
途中でユーザーへの質問・確認を挟まず、自律的に実行する。
処理中にエラーや異常が発生した場合は、その時点で処理を中止し、エラー内容を出力する。
正常に完了した場合は、実行結果を出力する。
sub-inquiry-urg-ntfは、サブエージェント2の名前です。
skill-inquiry-urg-ntfは、スキル2の名前です。
スキル2の中身
緊急性の有る問い合わせを社内通知する手順が書かれています。
スキル2「skill-inquiry-urg-ntf」のSKILL.md
---
name: skill-inquiry-urg-ntf
description: 緊急性ありの問い合わせを社内通知する。
---
# 役割
緊急性ありの問い合わせを社内通知する。
## 手順
1. ステップ1：緊急性が有る問い合わせのうち、社内未通知のものを、社内に通知
2. ステップ2：結果報告
各ステップの詳細は以下の通り。
## ステップ1：緊急性が有る問い合わせのうち、社内未通知のものを、社内に通知
緊急性が有る問い合わせのうち、社内未通知のものを、社内に通知する。
## ステップ2：結果報告
以下を出力する。
- 「緊急を要する問い合わせの社内通知が完了しました。」という文言
- ステップ1の結果（社内に通知した全問い合わせ番号）
- ※通知した問い合わせが1件もなければ「緊急を要する未通知の問い合わせはありませんでした。」という文言
※スキル2のSKILL.md内でも、MCPサーバーへの言及がないことについて
スキル1と同様、スキル2のSKILL.mdの中でも、MCPサーバーについての言及は全くありません。
このスキル2では、DBアクセスとSlack送信を行います。
にもかかわらず、SKILL.mdでは、両者を担当するMCPサーバー3に関することや、MCPツール名などは、一切書かれていません。
スキル1でも書きましたが、Claudeは、PJ起動時に既にMCPサーバーの情報を取得済みなので、自然言語での指示だけで適切なMCPサーバーとそのツールを自力で選択できるのです。
このスキル2での処理は非常にシンプルで、MCPサーバー3の内部構成もシンプルなので、Claudeがその選択に迷うことはないはずです。
例を挙げます。
スキルのSKILL.md内の
緊急性が有る問い合わせのうち、社内未通知のものを、社内に通知する。
を実際に行うのは、
MCPサーバー3inquiry-urg_ntfの、
ツールnotify_unnotified_urgent_inquiries
です。
このMCPサーバーとこのツールの機能説明文は、Claude Codeに事前に登録されています。
Claude Codeに事前に登録されている、ツール「notify_unnotified_urgent_inquiries」の機能説明文
緊急性が有る問い合わせのうち、まだ社内通知していないものを、社内に通知する。
そして、当該の問い合わせを社内通知済み状態に更新する。
（以下省略）
Claudeは、AIエージェント実行時には既にこの登録内容を知っているので、
「緊急性が有る問い合わせのうち、社内未通知のものを、社内に通知する。」
という自然言語での指示だけで、このツールを選択し、実行することができます。
これが成り立つのは、スキル1の場合と同じ理由です。
しかもスキル2は、処理もMCPサーバーの内部構成も、スキル1よりさらに単純です。
複雑な処理の場合や、MCPサーバーの役割分担が曖昧になる場合（複数の似た役割のMCPサーバー群、似た役割のツール群など）は、SKILL.md内での明示的な言及が必要になるでしょう。
AIエージェントを働かせてみる
作ったAIエージェントを働かせてみます。
実際に何回か定期実行させて、
どのような問い合わせに、どのような”対応案”が付加されたのかを、
問い合わせを何件かピックアップして、見てみます。
ケース1：ルール通りの回答をするだけの平易な質問
問い合わせの内容
| 問い合わせ・注文項目 | 内容 |
|---|---|
| 問い合わせ番号 | I000059 |
| 問い合わせ日 | 2026/3/31 |
| 問い合わせした顧客 | 山本 彩 |
| 注文したもの | 低糖質プロテインパン |
| 発送状態 | 未発送 |
| 問い合わせ本文 | （下記） |
問い合わせ本文
注文した低糖質プロテインパンをキャンセルできるか確認したいです。
AIの対応案
対応案策定で、AI（サブエージェント1）は、まず以下の2項目
・問い合わせ本文の要約
・対応ルート（人間が対応 or 自動送信メールで回答）
を他に先んじて決定するのでした。
本件は、注文したものがまだキャンセルできるか？という問い合わせです。
この問い合わせに対する答えは、ショッピングガイドに、そのまま載っています。
ショッピングガイドの「キャンセル」記述部分
## キャンセル
- 発送されるまでは無料でキャンセル可能です。発送後はキャンセルできません。
- キャンセル料は発生しません。
この問い合わせの対象である注文データ上では、まだ発送前です。
なので、ルール上、キャンセル可能、ということになります。
とてもシンプルな話で、それ以上は何もありません。
よって、
AIが回答文案を作成し、後に自動送信メールでの回答がなされる、
という流れになります。
ですので、上記2項目のAIの決定は、
| AI決定項目 | AI決定内容 |
|---|---|
| 問い合わせ本文の要約 | 注文した低糖質プロテインパン（注文番号O000343）のキャンセル可否を確認する問い合わせ。 |
| 対応ルート | 自動送信メールで回答 |
となりました。
対応ルートが「自動送信メールで回答」の場合、以降AIが決定する項目は「自動送信メールの回答の本文案」のみです。
AIは、以下のような回答文案を創出しました。
自動送信メールの回答の本文案 AI決定内容
お問い合わせいただきました「低糖質プロテインパン」（ご注文番号：O000343、数量4個）につきまして、
ご案内いたします。
本ご注文は現時点で発送前のため、キャンセルが可能でございます。キャンセル料は一切かかりません。
なお、キャンセルは発送前に限り承っており、発送後はお受けできませんのでご了承ください。
本ご注文はお届け予定日が2026年4月3日で、まもなく発送手配に入りますので、
キャンセルをご希望の場合はお早めにお申し付けください。
また、商品や数量の変更をご希望の場合も、
いったんキャンセルいただいたうえで改めてご注文いただく形となります。
対応ルートが「自動送信メールで回答」の場合、緊急性の有無の決定はされません。
＜ AIが策定した対応案 全項目 ＞
| AI決定項目 | AI決定内容 |
|---|---|
| 問い合わせ本文の要約 | 注文した低糖質プロテインパン（注文番号O000343）のキャンセル可否を確認する問い合わせ。 |
| 対応ルート | 自動送信メールで回答 |
| 自動送信メールの回答の本文案 |
お問い合わせいただきました「低糖質プロテインパン」（ご注文番号：O000343、数量4個）につきまして、 ご案内いたします。 本ご注文は現時点で発送前のため、キャンセルが可能でございます。キャンセル料は一切かかりません。 なお、キャンセルは発送前に限り承っており、発送後はお受けできませんのでご了承ください。 本ご注文はお届け予定日が2026年4月3日で、まもなく発送手配に入りますので、 キャンセルをご希望の場合はお早めにお申し付けください。 また、商品や数量の変更をご希望の場合も、 いったんキャンセルいただいたうえで改めてご注文いただく形となります。 |
★AIの効用
FAQや本稿PJのショッピングガイドのような公開情報に「まんま」載っていることを、わざわざていねいな文章を綴って回答するのは、人的リソースのムダです。
また、必須ではないが有用な周辺情報（本件で言うなら注文内容やお届け予定日）を回答に盛り込む場合、それらをいちいち調べなければなりません。
本件において、AIはそこらへんを丸ごと面倒を見てくれており、地味ですが確実に省力化されました。
ケース2：平易な質問ではあるが人間が対応すべきもの
問い合わせの内容
| 問い合わせ・注文項目 | 内容 |
|---|---|
| 問い合わせ番号 | I000051 |
| 問い合わせ日 | 2026/3/26 |
| 問い合わせした顧客 | 小林 恒一 |
| 注文したもの | マルチビタミン＆ミネラル |
| 問い合わせ本文 | （下記） |
問い合わせ本文
届いたマルチビタミン＆ミネラルは、朝と夜のどちらに飲めばよいですか。
AIの対応案
対応案策定で、AI（サブエージェント1）は、まず以下の2項目
・問い合わせ本文の要約
・対応ルート（人間が対応 or 自動送信メールで回答）
を他に先んじて決定します。
本件は、商品の使用法についての問い合わせです。
この問い合わせに対する答えは、商品マスタにも、その他のどこにもありません。
よって、AIが回答文案を作成し、後に自動送信メールで回答、という流れは不可能です。
人間が対応するほかはないはずです。
ですので、上記2項目のAIの決定は、
| AI決定項目 | AI決定内容 |
|---|---|
| 問い合わせ本文の要約 | 届いたマルチビタミン&ミネラルの服用タイミング（朝か夜か）を知りたい。 |
| 対応ルート | 人間が対応 |
となりました。
対応ルートが「人間が対応」の場合、以降AIが決定する項目は
「対応のヒント」「緊急性の有無」
の2つです。
「対応のヒント」は、問い合わせ対応案の全項目の中で核となる項目です。
対応方針と、その根拠となる様々な情報が盛り込まれます。
AIが本件に対して示した「対応のヒント」は、以下の通りです。
対応のヒント AI決定内容
【問い合わせ種別】
商品の使用方法（服用タイミング）に関する質問。
クレームではない。
【顧客情報】
小林恒一様（C007）。
購入額45,100円は全顧客平均33,517円の約1.35倍でやや上客。
過去に誤配送（I000010, 2026-01-23, 無添加アーモンドミルク）と
配送遅延（I000025, 2026-02-11, コラーゲンペプチドパウダー）
のクレームが各1件あり、いずれも解決済み。
直近はキャンセル確認（I000043）や保存方法の質問（I000048）など軽微な問い合わせのみ。
【同商品の類似問い合わせ先例】
マルチビタミン&ミネラルの飲むタイミングに関する問い合わせは
I000027（Emily Johnson, 2026-02-19）および
I000040（吉田翔, 2026-03-02）で先例があり、
いずれも「食後の服用」を一般的に案内する対応で完了している。
【対応方針】
先例に倣い、飲むタイミングの一般的な案内（食後等）を回答すれば完了。
過去にトラブルが2件あった顧客のため、丁寧な対応を心がけること。
【問い合わせ種別】
商品の使用法の質問であり、クレームではない。
→
後の「緊急性の有無」の決定要素の1つになります。
【顧客情報】
この顧客との関係性の1つである「手厚く扱うべき上客か」を、販売統計を見て判断しています。
この顧客への販売実績は、全顧客平均の1.35倍であり、”やや上客”にすぎない。
→
対応に際し、特別扱い不要、ということです。
【顧客情報】
この顧客との関係性のもう1つである「ケアすべき客か」を、過去のこの顧客とのやりとりを収集して、判断しています。
解決済みではあるものの、過去に2回ほど迷惑をかけている。
→
ケアすべき客、という判断がされたようです。
【同商品の類似問い合わせ先例】
同商品に対する同じ問い合わせが過去に2回あり、先行事例として利用可能。
→
それらと同じ回答をする、という判断がされました。
⇒
これらを盛り込んで
【対応方針】
先例に倣い、飲むタイミングの一般的な案内（食後等）を回答すれば完了。
過去にトラブルが2件あった顧客のため、丁寧な対応を心がけること。
－
という対応方針をAIが決定しました。
次は、「緊急性の有無」の決定です。
上記「対応のヒント」で明らかにしたことが、決定要素として使用されています。
・単なる商品の使用法の質問である（上記「対応のヒント」参照）
・特急対応すべき上客ではない（上記「対応のヒント」参照）
・過去、多大な迷惑をかけた客ではない（上記「対応のヒント」参照）
⇒
AIは、緊急性無し、と決定しました。
＜ AIが策定した対応案 全項目 ＞
| AI決定項目 | AI決定内容 |
|---|---|
| 問い合わせ本文の要約 | 届いたマルチビタミン&ミネラルの服用タイミング（朝か夜か）を知りたい。 |
| 対応ルート | 人間が対応 |
| 対応のヒント |
【問い合わせ種別】 商品の使用方法（服用タイミング）に関する質問。 クレームではない。 【顧客情報】 小林恒一様（C007）。 購入額45,100円は全顧客平均33,517円の約1.35倍でやや上客。 過去に誤配送（I000010, 2026-01-23, 無添加アーモンドミルク）と 配送遅延（I000025, 2026-02-11, コラーゲンペプチドパウダー） のクレームが各1件あり、いずれも解決済み。 直近はキャンセル確認（I000043）や保存方法の質問（I000048）など軽微な問い合わせのみ。 【同商品の類似問い合わせ先例】 マルチビタミン&ミネラルの飲むタイミングに関する問い合わせは I000027（Emily Johnson, 2026-02-19）および I000040（吉田翔, 2026-03-02）で先例があり、 いずれも「食後の服用」を一般的に案内する対応で完了している。 【対応方針】 先例に倣い、飲むタイミングの一般的な案内（食後等）を回答すれば完了。 過去にトラブルが2件あった顧客のため、丁寧な対応を心がけること。 |
| 緊急性の有無 | 無し |
★AIの効用
本件は、軽微な質問です。
もしAIでなく人力で対応した場合、コスパ意識から、AIほどにはデータ収集をしなかったんじゃないか、とも思います。
その結果、過去に2回迷惑をかけた顧客である、という認識がないまま、フツーの対応をしてしまった可能性があります。
そういう意味では、省力化だけでなく品質向上の効用もありました。
ケース3：普通のクレーム
問い合わせの内容
| 問い合わせ・注文項目 | 内容 |
|---|---|
| 問い合わせ番号 | I000053 |
| 問い合わせ日 | 2026/3/26 |
| 問い合わせした顧客 | 伊藤 健 |
| 注文したもの | 無添加アーモンドミルク |
| 問い合わせ本文 | （下記） |
問い合わせ本文
届いた無添加アーモンドミルクの容器がへこんでいて、少し漏れていました。交換してください。
AIの対応案
対応案策定で、AI（サブエージェント1）は、まず以下の2項目
・問い合わせ本文の要約
・対応ルート（人間が対応 or 自動送信メールで回答）
を他に先んじて決定します。
本件は、「商品が破損して被害が出ている。交換せよ！」というクレームです。
当然、人間が対応しなければならないはずです。
ですので、上記2項目のAIの決定は、
| AI決定項目 | AI決定内容 |
|---|---|
| 問い合わせ本文の要約 | 2026年3月27日に納品された「無添加アーモンドミルク」（注文番号O000326・3個）について、 容器がへこんでおり中身が少し漏れていたとの申し出。 交換を希望している。 商品破損の訴え。 |
| 対応ルート | 人間が対応 |
となりました。
対応ルートが「人間が対応」の場合、以降AIが決定する項目は
「対応のヒント」「緊急性の有無」
の2つです。
「対応のヒント」は、問い合わせ対応案の全項目の中で核となる項目です。
対応方針と、その根拠となる様々な情報が盛り込まれます。
AIが本件に対して示した「対応のヒント」は、以下の通りです。
対応のヒント AI決定内容
【問い合わせ内容】
2026/3/24注文・3/27納品の「無添加アーモンドミルク」（注文番号O000326・3個）
について、容器がへこんで中身が少し漏れていたという破損の申し出。
交換を希望している。
【顧客の状況】
伊藤健様（C005）。
累計購入額28,030円で、全顧客平均33,517円の約0.84倍と平均をやや下回る一般顧客。
トラブル歴は2026/1/28の納期遅延（肝機能サポートサプリ）1件のみ。
以降は納品予定日の確認やキャンセル可否の問い合わせに自動回答で対応しており、関係は概ね良好。
今回が2件目のトラブルにあたる。
【参考にすべき先行対応事例】
・I000037（2026/2/28 渡辺直子様）：
「容器がへこんでいて少し漏れていた」という今回とほぼ同一の申し出。
新品の交換発送を迅速に手配し、お詫びと到着予定日をメールで丁寧に案内。
同一商品での容器破損2件目だったため、梱包強化の必要性を商品担当に共有した。
・I000015（2026/1/26 John Smith様）：
破損品について「返品不要」である旨をあわせて案内している。
・I000014・I000022・I000033：
破損・梱包不良は一貫して「即時交換発送 → お詫び → 商品担当への共有」で対応。
【対応方針】
新品の無添加アーモンドミルクの交換発送を迅速に手配し、到着予定日をメールで案内する。
破損品の返品は不要である旨も明示する。
中身が漏れており衛生面のご不安が想定されるため、
破損した商品はお召し上がりにならないようご案内する。
初回のクレーム対応（1/28の遅延）に続く2件目であることを踏まえ、お詫びを丁寧に添える。
無添加アーモンドミルクでの容器破損報告は初めてだが、
飲料カテゴリでの液漏れ・容器破損は他商品でも継続的に発生しているため、
梱包品質の問題として商品担当に共有する。
【問い合わせ内容】
商品破損と被害の発生のクレーム。
→
対応全般に影響するとともに、後の「緊急性の有無」の決定要素の1つになります。
【顧客情報】
この顧客との関係性の1つである「手厚く扱うべき上客か」を、販売統計を見て判断しています。
この顧客への販売実績は、全顧客平均の約0.84倍でしかない。
→
対応に際し、特別扱い不要、ということです。
【顧客情報】
この顧客との関係性のもう1つである「ケアすべき客か」を、過去のこの顧客とのやりとりを収集して、判断しています。
迷惑をかけたのは、解決済みの配送遅延が1回だけ。
以降は納品予定日の確認やキャンセル可否の問い合わせに自動回答で対応。
→
関係は概ね良好、特にケア不要の客、という判断がされたようです。
【参考にすべき先行対応事例】
類似事案に対し、過去、以下のような対応がされたことを提示しています。
・同一商品での容器破損2件目になった時点で、梱包強化の必要性を商品担当に共有
・破損品について「返品不要」である旨をあわせて案内
・破損・梱包不良は一貫して「即時交換発送 → お詫び → 商品担当への共有」で対応
⇒
これらを盛り込んで
【対応方針】
新品の無添加アーモンドミルクの交換発送を迅速に手配し、到着予定日をメールで案内する。
破損品の返品は不要である旨も明示する。
中身が漏れており衛生面のご不安が想定されるため、破損した商品はお召し上がりにならないようご案内する。
初回のクレーム対応（1/28の遅延）に続く2件目であることを踏まえ、お詫びを丁寧に添える。
無添加アーモンドミルクでの容器破損報告は初めてだが、
飲料カテゴリでの液漏れ・容器破損は他商品でも継続的に発生しているため、梱包品質の問題として商品担当に共有する。
－
という対応方針をAIが決定しました。
次は、「緊急性の有無」の決定です。
商品破損・被害発生というクレームである
⇒
AIは、緊急性有り、と決定しました。
＜ AIが策定した対応案 全項目 ＞
| AI決定項目 | AI決定内容 |
|---|---|
| 問い合わせ本文の要約 | 2026年3月27日に納品された「無添加アーモンドミルク」（注文番号O000326・3個） について、容器がへこんでおり中身が少し漏れていたとの申し出。 交換を希望している。 商品破損の訴え。 |
| 対応ルート | 人間が対応 |
| 対応のヒント |
【問い合わせ内容】 2026/3/24注文・3/27納品の「無添加アーモンドミルク」（注文番号O000326・3個）について、容器がへこんで中身が少し漏れていたという破損の申し出。 交換を希望している。 【顧客情報】 伊藤健様（C005）。 累計購入額28,030円で、全顧客平均33,517円の約0.84倍と平均をやや下回る一般顧客。 トラブル歴は2026/1/28の納期遅延（肝機能サポートサプリ）1件のみ。 以降は納品予定日の確認やキャンセル可否の問い合わせに自動回答で対応しており、関係は概ね良好。 今回が2件目のトラブルにあたる。 【参考にすべき先行対応事例】 ・I000037（2026/2/28 渡辺直子様）： 「容器がへこんでいて少し漏れていた」という今回とほぼ同一の申し出。 新品の交換発送を迅速に手配し、お詫びと到着予定日をメールで丁寧に案内。 同一商品での容器破損2件目だったため、梱包強化の必要性を商品担当に共有した。 ・I000015（2026/1/26 John Smith様）： 破損品について「返品不要」である旨をあわせて案内している。 ・I000014・I000022・I000033： 破損・梱包不良は一貫して「即時交換発送 → お詫び → 商品担当への共有」で対応。 【対応方針】 新品の無添加アーモンドミルクの交換発送を迅速に手配し、到着予定日をメールで案内する。 破損品の返品は不要である旨も明示する。 中身が漏れており衛生面のご不安が想定されるため、破損した商品はお召し上がりにならないようご案内する。 初回のクレーム対応（1/28の遅延）に続く2件目であることを踏まえ、お詫びを丁寧に添える。 無添加アーモンドミルクでの容器破損報告は初めてだが、 飲料カテゴリでの液漏れ・容器破損は他商品でも継続的に発生しているため、梱包品質の問題として商品担当に共有する。 |
| 緊急性の有無 | 有り |
Slackによる社内通知
緊急性有り、と決定されたので、Slackによる社内通知の対象となります。
サブエージェント2が、本件を含め、緊急性が有る全ての問い合わせを一括してSlackの社内専用チャンネルに送信します。
[緊急性有り → Slackの社内専用チャンネルに一括送信]
★AIの効用
破損・梱包不良という類似事案には、
「即時交換発送 → お詫び → 商品担当への共有」で対応、
破損品の返品は不要、と明示して案内、
という先例を提示しています。
この情報により、対応担当者は、ここから思考を開始することができます。
「飲料カテゴリでの液漏れ・容器破損は他商品でも継続的に発生しているため、梱包品質の問題として商品担当に共有」というAIの「社内」対応方針が示されました。
人間にとって、問い合わせ対応は、問い合わせてきた顧客に対する「社外」対応、が1番に来ます。
もしAIでなく人間が本件に対応したとしたら、上記の「社内」対応という発想をしなかった可能性があります。
これも1つの品質向上と言えます。
ケース4：重篤なクレーム
問い合わせの内容
| 問い合わせ・注文項目 | 内容 |
|---|---|
| 問い合わせ番号 | I000062 |
| 問い合わせ日 | 2026/4/2 |
| 問い合わせした顧客 | 田中 裕二 |
| 注文したもの | 発酵黒にんにく粒 |
| 問い合わせ本文 | （下記） |
問い合わせ本文
発酵黒にんにく粒は4月2日に届くはずでしたが、もう夜なのに、また届いていません。
これまで何度も納期遅れがあり、さすがに我慢できません。
至急、どうなっているのか説明してください。
AIの対応案
対応案策定で、AI（サブエージェント1）は、まず以下の2項目
・問い合わせ本文の要約
・対応ルート（人間が対応 or 自動送信メールで回答）
を他に先んじて決定します。
本件は、納期遅れに対するクレームです。
が、度重なる納期遅れに怒り心頭、「責任者出てこい！」、という状態です。
当然、人間が対応しなければならないはずです。
ですので、上記2項目のAIの決定は、
| AI決定項目 | AI決定内容 |
|---|---|
| 問い合わせ本文の要約 | 発酵黒にんにく粒が4月2日の納品予定日を過ぎても届かない。 過去に何度も納期遅れがあり我慢の限界と表明。 至急の状況説明を要求。 |
| 対応ルート | 人間が対応 |
となりました。
対応ルートが「人間が対応」の場合、以降AIが決定する項目は
「対応のヒント」「緊急性の有無」
の2つです。
「対応のヒント」は、問い合わせ対応案の全項目の中で核となる項目です。
対応方針と、その根拠となる様々な情報が盛り込まれます。
AIが本件に対して示した「対応のヒント」は、以下の通りです。
対応のヒント AI決定内容
【問い合わせ種別 - 最重要案件】
納期遅延の訴え。
発酵黒にんにく粒が4月2日の予定を過ぎても届かず、「さすがに我慢できません」と強い怒りを表明。
至急の説明を要求。
注文データ上、2026-03-31に発送済みだが未納品。
【顧客情報 - 最大級の注意】
田中裕二様（C011）。
購入額42,510円は全顧客平均35,195円の約1.21倍でやや上客。
この顧客は過去に配送遅延のクレームを4回出しており、すべて実際の遅延が確認されている：
(1) I000002（2026-01-05, 乳酸菌ドリンクEX, 4日遅延）、
(2) I000008（2026-01-22, 発酵黒にんにく粒, 4日遅延, 怒りの表明）、
(3) I000023（2026-02-10, オーガニック玄米ブレンド, 4日遅延）、
(4) I000044（2026-03-13, 発酵黒にんにく粒, 4日遅延, エスカレーション対応済み）。
本件で5回目。
前回I000044ではエスカレーションし配送業者との調整を行ったにもかかわらず、再び同じ問題が発生している。
【同商品の過去トラブル】
発酵黒にんにく粒では
配送遅延（I000008 田中裕二、I000044 田中裕二）、
誤配送（I000030 加藤優子、I000047 Emily Johnson）
と複数のトラブルが報告されている。
【対応方針 - 最優先】
5回目の配送遅延であり、前回のエスカレーション後も改善されていない最悪の状況。
定型対応は絶対に避け、上位管理者も含めた対応が必要。
配送状況の即時確認、電話での直接のお詫び、具体的かつ実効性のある再発防止策の提示が不可欠。
顧客離反のリスクが極めて高い。
【問い合わせ種別 - 最重要案件】
納期遅延。強い怒りを表明。
→
最重要案件。
【顧客情報 - 最大級の注意】
この顧客への販売実績は全顧客平均の約1.21倍で、”やや上客”である。しかし本件ではそれよりも重要なことがある。
この顧客に対して、過去4回も同じ納期遅延を発生させており、多大な迷惑をかけている。
しかも、直近4回目では、配送業者にエスカレーションして対策したのにもかかわらず、効果なし。
→
最大級の注意を払うべき顧客。
【同商品の過去トラブル】
この商品は、過去、配送遅延2回と誤配送2回の計4回の配送トラブルが生じている。
→
具体的かつ実効性のある再発防止策の提示が不可欠。
⇒
これらを盛り込んで
【対応方針 - 最優先】
5回目の配送遅延であり、前回のエスカレーション後も改善されていない最悪の状況。
定型対応は絶対に避け、上位管理者も含めた対応が必要。
配送状況の即時確認、電話での直接のお詫び、具体的かつ実効性のある再発防止策の提示が不可欠。
顧客離反のリスクが極めて高い。
－
という最優先の対応方針をAIが決定しました。
次は、「緊急性の有無」の決定です。
5回目の納期遅延で、顧客が正当に怒っている。
⇒
AIは、緊急性有り、と決定しました。
＜ AIが策定した対応案 全項目 ＞
| AI決定項目 | AI決定内容 |
|---|---|
| 問い合わせ本文の要約 |
発酵黒にんにく粒が4月2日の納品予定日を過ぎても届かない。 過去に何度も納期遅れがあり我慢の限界と表明。 至急の状況説明を要求。 |
| 対応ルート | 人間が対応 |
| 対応のヒント |
【問い合わせ種別 - 最重要案件】 納期遅延の訴え。 発酵黒にんにく粒が4月2日の予定を過ぎても届かず、「さすがに我慢できません」と強い怒りを表明。 至急の説明を要求。 注文データ上、2026-03-31に発送済みだが未納品。 【顧客情報 - 最大級の注意】 田中裕二様（C011）。 購入額42,510円は全顧客平均35,195円の約1.21倍でやや上客。 この顧客は過去に配送遅延のクレームを4回出しており、すべて実際の遅延が確認されている： (1) I000002（2026-01-05, 乳酸菌ドリンクEX, 4日遅延）、 (2) I000008（2026-01-22, 発酵黒にんにく粒, 4日遅延, 怒りの表明）、 (3) I000023（2026-02-10, オーガニック玄米ブレンド, 4日遅延）、 (4) I000044（2026-03-13, 発酵黒にんにく粒, 4日遅延, エスカレーション対応済み）。 本件で5回目。 前回I000044ではエスカレーションし配送業者との調整を行ったにもかかわらず、再び同じ問題が発生している。 【同商品の過去トラブル】 発酵黒にんにく粒では 配送遅延（I000008 田中裕二、I000044 田中裕二）、 誤配送（I000030 加藤優子、I000047 Emily Johnson） と複数のトラブルが報告されている。 【対応方針 - 最優先】 5回目の配送遅延であり、前回のエスカレーション後も改善されていない最悪の状況。 定型対応は絶対に避け、上位管理者も含めた対応が必要。 配送状況の即時確認、電話での直接のお詫び、具体的かつ実効性のある再発防止策の提示が不可欠。 顧客離反のリスクが極めて高い。 |
| 緊急性の有無 | 有り |
Slackによる社内通知
緊急性有り、と決定されたので、Slackによる社内通知の対象となります。
サブエージェント2が、本件を含め、緊急性が有る全ての問い合わせを一括してSlackの社内専用チャンネルに送信します。
[緊急性有り → Slackの社内専用チャンネルに一括送信]
★AIの効用
クレームを送ってきたこの顧客「田中 裕二」氏がカンカンに怒っているのは文面を見て分かるのですが、そこに至るまでの諸々の経緯、この商品自体のトラブルの多さといった、対応に際し価値のある情報を揃えておいてくれるのは、AIならでは、でしょう。
ケース5：平易な質問なのに緊急対応
問い合わせの内容
| 問い合わせ・注文項目 | 内容 |
|---|---|
| 問い合わせ番号 | I000055 |
| 問い合わせ日 | 2026/3/30 |
| 問い合わせした顧客 | 田中 裕二 |
| 注文したもの | 高カカオポリフェノールチョコ |
| 問い合わせ本文 | （下記） |
問い合わせ本文
届いた高カカオポリフェノールチョコは、1日にどのくらい食べればよいですか。
AIの対応案
対応案策定で、AI（サブエージェント1）は、まず以下の2項目
・問い合わせ本文の要約
・対応ルート（人間が対応 or 自動送信メールで回答）
を他に先んじて決定します。
本件は、商品の使用法についての問い合わせです。
この問い合わせに対する答えは、商品マスタにも、その他のどこにもありません。
よって、AIが回答文案を作成し、後に自動送信メールで回答、という流れは不可能です。
が、それ以上に大事なのは、
問い合わせした顧客が、
過去に納期遅延を4回も発生させて多大な迷惑をかけている「田中 裕二」氏である、
ということです（ケース4参照）。
これは、人間が対応する必要があるはずです。
ですので、上記2項目のAIの決定は、
| AI決定項目 | AI決定内容 |
|---|---|
| 問い合わせ本文の要約 | 届いた高カカオポリフェノールチョコの1日あたりの適量（摂取量の目安）を知りたい。 |
| 対応ルート | 人間が対応 |
となりました。
対応ルートが「人間が対応」の場合、以降AIが決定する項目は
「対応のヒント」「緊急性の有無」
の2つです。
「対応のヒント」は、問い合わせ対応案の全項目の中で核となる項目です。
対応方針と、その根拠となる様々な情報が盛り込まれます。
AIが本件に対して示した「対応のヒント」は、以下の通りです。
対応のヒント AI決定内容
【問い合わせ種別】
商品の使用方法（1日の摂取量の目安）に関する質問。
クレームではない。
【顧客情報 - 要注意】
田中裕二様（C011）。
購入額42,510円は全顧客平均34,947円の約1.22倍でやや上客。
過去に配送遅延のクレームを4回出しており、すべて実際の遅延が確認されている：
(1) I000002（2026-01-05, 乳酸菌ドリンクEX）、
(2) I000008（2026-01-22, 発酵黒にんにく粒, 怒りの表明あり）、
(3) I000023（2026-02-10, オーガニック玄米ブレンド, 度重なる遅延への不満）、
(4) I000044（2026-03-13, 発酵黒にんにく粒, 4度目の遅延、エスカレーション対応済み）。
当社がこの顧客に多大な迷惑をかけ続けている。
【同商品の先例】
高カカオポリフェノールチョコへの問い合わせは今回が初。類似の先例なし。
【対応方針】
1日の摂取量の目安を回答すること。
クレームではないが、過去に配送遅延で多大な迷惑をかけている顧客であるため、
丁寧かつ迅速に対応し、顧客満足度の回復につなげること。
好印象を与える対応を心がけるべき。
【問い合わせ種別】
単なる商品の使用法についての質問であり、クレームではない。
【顧客情報 - 要注意】
この顧客への販売実績は全顧客平均の約1.22倍で、”やや上客”だが、そこは今重要ではない。
この顧客に対して、過去4回も納期遅延を発生させており、多大な迷惑をかけている。
→
対応に際し要注意となる顧客。
【同商品の先例】
過去事例が無く、参考になる先例無し。
⇒
これらを盛り込んで
【対応方針】
1日の摂取量の目安を回答すること。
クレームではないが、過去に配送遅延で多大な迷惑をかけている顧客であるため、
丁寧かつ迅速に対応し、顧客満足度の回復につなげること。
好印象を与える対応を心がけるべき。
－
という対応方針をAIが決定しました。
次は、「緊急性の有無」の決定です。
単なる商品の使用法の質問ではあるが、多大な迷惑をかけている顧客である。
⇒
AIは、緊急性有り、と決定しました。
＜ AIが策定した対応案 全項目 ＞
| AI決定項目 | AI決定内容 |
|---|---|
| 問い合わせ本文の要約 | 届いた高カカオポリフェノールチョコの1日あたりの適量（摂取量の目安）を知りたい。 |
| 対応ルート | 人間が対応 |
| 対応のヒント |
【問い合わせ種別】 商品の使用方法（1日の摂取量の目安）に関する質問。 クレームではない。 【顧客情報 - 要注意】 田中裕二様（C011）。 購入額42,510円は全顧客平均34,947円の約1.22倍でやや上客。 過去に配送遅延のクレームを4回出しており、すべて実際の遅延が確認されている： (1) I000002（2026-01-05, 乳酸菌ドリンクEX）、 (2) I000008（2026-01-22, 発酵黒にんにく粒, 怒りの表明あり）、 (3) I000023（2026-02-10, オーガニック玄米ブレンド, 度重なる遅延への不満）、 (4) I000044（2026-03-13, 発酵黒にんにく粒, 4度目の遅延、エスカレーション対応済み）。 当社がこの顧客に多大な迷惑をかけ続けている。 【同商品の先例】 高カカオポリフェノールチョコへの問い合わせは今回が初。類似の先例なし。 【対応方針】 1日の摂取量の目安を回答すること。 クレームではないが、過去に配送遅延で多大な迷惑をかけている顧客であるため、 丁寧かつ迅速に対応し、顧客満足度の回復につなげること。 好印象を与える対応を心がけるべき。 |
| 緊急性の有無 | 有り |
Slackによる社内通知
緊急性有り、と決定されたので、Slackによる社内通知の対象となります。
サブエージェント2が、本件を含め、緊急性が有る全ての問い合わせを一括してSlackの社内専用チャンネルに送信します。
[緊急性有り → Slackの社内専用チャンネルに一括送信]
★AIの効用
本件は、外面上は、商品の使用法の質問という単純な問い合わせです。
本件に対し、AIではなく人力で対応した場合、「田中 裕二」氏は最もケアすべき顧客である、と知っている人間ではない限り、フツーのスピードで対応した可能性があります。
それは、きっとダメージは生じなかったでしょうが、同時に、顧客満足度の回復につながることもなかったでしょう。
AIが、過去データをもとに、「田中 裕二」氏は最もケアすべき顧客であるから緊急性の有る問い合わせとして扱う、という判断をしたからこそ、迅速な対応ができました。
顧客満足度の回復につながる芽は摘まれなかった、ということです。
おわりに
本稿のAIエージェントは、大したことをしていません。
適切と思われるデータを取ってきて、読んで、答えを考え、それを書き込む。
急ぐべき、と思ったら「急げ！」と叫ぶ。
ただそれを繰り返しているだけです。
そして、そこでAIが読んだデータは、AIのために用意したものではありません。
販売実績も、問い合わせ対応履歴も、業務のために元々あったシステム資産です。
それらをAIが読める状態にしただけです。
それでも、対応の省力化と品質向上につながりました。
同じことは、他の業務でも成り立つはずです。
また、既存のシステム資産は、データだけではありません。
長年動かしてきたシステムには、多様な業務ルールを実行する機能が積み上がっています。
それらをMCPサーバー化するなどしてAIから使えるようにすれば、できることはもっと増えるはずです。
【参考】DBとMCPサーバー
ここからは、本稿で使用したDBとMCPサーバー群の中身についての参考情報です。
技術的な内容なので、興味のない方は読み飛ばして構いません。
DB
Supabase上で構築しました。
マスタ（顧客、商品、社員）と、業務データのテーブルがあります。
また、それらのデータを取得・集計・更新するFunction群があります。
DB外部からテーブルへのアクセスは、全てそれらFunction経由で行われます。
クリックして表示
構成
以下のテーブルとFunctionによって構成されています。
| 種類 | 名称 | 内容・機能 |
|---|---|---|
| Table | CUTOMER | 顧客マスタ。 |
| Table | ITEM | 商品マスタ。 |
| Table | ITEM_CATEGORY | 商品カテゴリーマスタ。商品のカテゴリーを定義。 |
| Table | MEMBER | 社員マスタ。 |
| Table | SALES | 販売データ。 |
| Table | ORDERS | 注文データ。 |
| Table | INQUIRY | 問い合わせデータ。 |
| Table | INQUIRY_RES | 問い合わせ対応データ。 |
| Function | get_customers | 顧客マスタのデータを検索して返す。 |
| Function | get_items | 商品マスタと商品カテゴリーマスタのデータを検索して返す。 |
| Function | get_members | 社員マスタのデータを検索して返す。 |
| Function | get_sales | 販売データを検索して返す。集計無し。販売データ1個＝戻り値1行。 |
| Function | get_sales_sum_by_customer_item_member | 販売データを検索し、顧客・商品・社員で集計して返す。 |
| Function | get_sales_sum_by_customer_itemcat_member | 販売データを検索し、顧客・商品カテゴリー・社員で集計して返す。 |
| Function | get_sales_sum_by_customer | 販売データを検索し、顧客で集計して返す。 同時に、全顧客間平均販売額も返す。 |
| Function | get_orders | 注文データを検索して返す。注文データ1個＝戻り値1行。 |
| Function | get_inquiries | 問い合わせデータを検索して返す。 戻り値1行は、 「1件の問い合わせ」＋「その問い合わせの対象である1件の注文」 という構成である。 どんな注文に対するどんな問い合わせなのか、戻り値の1行を見ればわかる。 |
| Function | get_inquiry_responses | 問い合わせ対応データを検索して返す。 戻り値1行は、 「1件の問い合わせ対応」＋「その対応をした1件の問い合わせ」＋「その問い合わせの対象である1件の注文」 という構成である。 どんな注文に対するどんな問い合わせで、それに対してどんな対応をしたのか、戻り値の1行を見ればわかる。 |
| Function | update_inquiry | 問い合わせデータ1個を更新する。 限られたフィールドのみ更新可。 |
| Function | get_unnotified_urgent_inquiries | 緊急性有りで社内未通知の問い合わせデータ全件を返す。 |
| Function | mark_unnotified_urgent_inquiries_as_notified | 緊急性有りで社内未通知の問い合わせデータ全件を、社内通知済みに一括更新する。 |
[テーブル]
MCPサーバー「sales_stat」
本稿の「MCPサーバー1」です。
主に、販売データの統計を返します。
また、マスタデータの取得も行います。
FastMCPを使用しています。
クリックして表示
| 項目 | 内容 |
|---|---|
| サーバー名 | sales_stat |
| ファイル名 | mcp_server_sales_stat.py |
| Transport | Streamable HTTP |
| 認証 | 静的Bearer Token |
認証に静的Bearer Tokenを用いています。
トークンが平文でHTTPリクエストに含まれるため、本来はHTTPSで通信すべきですが、閉じられた宅内LANで使用しているだけなので、そこまではしませんでした。
ツール
各ツール内部で、適したDBのFunctionを使用しています。
| 名称 | 機能 |
|---|---|
| get_customers | DBから顧客情報を検索して返す。 |
| get_items | DBから商品情報を検索して返す。 |
| get_members | DBから社員情報を検索して返す。 |
| get_sales | DBから販売データを検索して返す。集計無し。販売データ1個＝戻り値1行。 |
| get_sales_sum_by_customer_item_member | DBから顧客・商品・社員で集計された販売データを取得して返す。 |
| get_sales_sum_by_customer_itemcat_member | DBから顧客・商品カテゴリー・社員で集計された販売データを取得して返す。 |
| get_sales_sum_by_customer | DBから顧客で集計された販売データを取得して返す。 同時に、全顧客間平均販売額も返す。 |
コード
MCPサーバー「mcp_server_sales_stat.py」
from fastmcp import FastMCP
import os
from typing import Annotated, List
from pydantic import BaseModel, Field
from typing import Optional
import logging
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import AsyncIterator
import psycopg2
import psycopg2.extras
from pathlib import Path
from dotenv import load_dotenv
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier
load_dotenv(Path(__file__).resolve().parent / ".env")
# ロギング設定（出力フォーマット追加）
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [SALES-STAT-SERVER] %(message)s"
)
logger = logging.getLogger("fast-mcp-server-sales-stat")
# DB接続をグローバル変数として保持
# ※カーソルはtool呼び出しごとに新規作成する
db_connection = None
@dataclass
class AppContext:
server_name: str = "MCP Server for sales statistics"
start_time: str = ""
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
global db_connection
try:
yield AppContext()
finally:
# 終了時：クリーンアップ
if db_connection is not None:
db_connection.close()
db_connection = None
# verifierを作り、FastMCPに渡す
_bearer_token = os.environ.get("MCP_SALES_STAT_BEARER_TOKEN")
if not _bearer_token:
raise ValueError("環境変数 'MCP_SALES_STAT_BEARER_TOKEN' が設定されていません。")
verifier = StaticTokenVerifier(
tokens={
_bearer_token: {
"client_id": "sales-stat-client",
"scopes": ["read:data"]
}
},
required_scopes=["read:data"]
)
mcp = FastMCP("Health_food_sales_system_server",
instructions="健康食品の販売システムにおいて、販売データ、マスタデータの取得を行う。販売統計の取得も行う。",
lifespan=app_lifespan,
auth=verifier)
def get_db_connection():
"""
DB接続を返す。接続が無い or 切断済みの場合のみ新規作成。
カーソルは呼び出し元で都度作成すること。
"""
global db_connection
if db_connection is None or db_connection.closed:
try:
db_dsn = os.environ.get('DBDSN')
if not db_dsn:
raise ValueError("環境変数 'DBDSN' が設定されていません。")
# DSN (Data Source Name) を使用して接続
db_connection = psycopg2.connect(dsn=db_dsn, sslmode='require')
# autocommit モードを設定
db_connection.autocommit = True
except Exception as e:
raise
return db_connection
# ----------------------------------------------------------------
# get_customers ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class CustomerRecord(BaseModel):
"""顧客マスタの1レコード"""
id: int = Field(description="顧客ID（システム内部の一意な数値ID）")
code: str = Field(description="顧客コード（業務上の顧客識別コード）")
name: str = Field(description="顧客名")
address: Optional[str] = Field(description="顧客の住所。未登録の場合はnull。")
tel: Optional[str] = Field(description="顧客の電話番号。未登録の場合はnull。")
class CustomerSearchResult(BaseModel):
"""
get_customers ツールの戻り値。
count=0 かつ customers=[] の場合は検索条件に合致する顧客が0件であることを示す（正常応答）。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
count: int = Field(description="取得した顧客レコード数。0の場合は検索条件に合致する顧客なし。")
customers: List[CustomerRecord] = Field(
description="顧客レコードの一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# get_customers ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def get_customers(
code_customer: Annotated[Optional[str], Field(
description="顧客マスタの検索条件。顧客コードによる完全一致検索。オプション。",
default=None
)],
name_str_customer: Annotated[Optional[str], Field(
description="顧客マスタの検索条件。顧客名の一部または全部による部分一致検索。オプション。",
default=None
)]
) -> CustomerSearchResult:
"""
健康食品の販売システムにおいて、顧客マスタの検索を行い、顧客データを返す。
引数をすべて省略した場合は全件取得となる。
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
# カーソルはこのtool呼び出しのスコープ内で作成・クローズする
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# DBのFunctionを呼び出す
cur.execute(
"SELECT * FROM public.get_customers(%s, %s)",
(code_customer, name_str_customer)
)
rows = cur.fetchall()
finally:
cur.close()
# RealDictRow → CustomerRecord に変換（0件の場合は空リストのまま返す）
customer_list = [
CustomerRecord(
id=row["id"],
code=row["code"],
name=row["name"],
address=row.get("address"),
tel=row.get("tel")
)
for row in rows
]
return CustomerSearchResult(
count=len(customer_list),
customers=customer_list
)
except Exception as e:
# raise することで FastMCP が isError=True のエラー応答を自動生成する
# （MCPプロトコル標準の方式）
logger.error(f"get_customers エラー: {e}")
raise
# ----------------------------------------------------------------
# get_items ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class ItemRecord(BaseModel):
"""商品マスタの1レコード"""
id: int = Field(description="商品ID（システム内部の一意な数値ID）")
code: str = Field(description="商品コード（業務上の商品識別コード）")
name: str = Field(description="商品名")
code_item_category: str = Field(description="商品カテゴリーのコード")
name_item_category: str = Field(description="商品カテゴリーの名前")
price: int = Field(description="商品の価格")
class ItemSearchResult(BaseModel):
"""
get_items ツールの戻り値。
count=0 かつ items=[] の場合は検索条件に合致する商品が0件であることを示す（正常応答）。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
count: int = Field(description="取得した商品レコード数。0の場合は検索条件に合致する商品なし。")
items: List[ItemRecord] = Field(
description="商品レコードの一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# get_items ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def get_items(
code_item: Annotated[Optional[str], Field(
description="商品マスタの検索条件。商品コードによる完全一致検索。オプション。",
default=None
)],
name_str_item: Annotated[Optional[str], Field(
description="商品マスタの検索条件。商品名の一部または全部による部分一致検索。オプション。",
default=None
)],
code_item_category: Annotated[Optional[str], Field(
description="商品マスタの検索条件。商品カテゴリーコードによる完全一致検索。オプション。",
default=None
)],
name_str_item_category: Annotated[Optional[str], Field(
description="商品マスタの検索条件。商品カテゴリー名の一部または全部による部分一致検索。オプション。",
default=None
)]
) -> ItemSearchResult:
"""
健康食品の販売システムにおいて、商品マスタの検索を行い、商品データを返す。
引数をすべて省略した場合は全件取得となる。
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
# カーソルはこのtool呼び出しのスコープ内で作成・クローズする
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# DBのFunctionを呼び出す
cur.execute(
"SELECT * FROM public.get_items(%s, %s, %s, %s)",
(code_item, name_str_item, code_item_category, name_str_item_category)
)
rows = cur.fetchall()
finally:
cur.close()
# RealDictRow → ItemRecord に変換（0件の場合は空リストのまま返す）
item_list = [
ItemRecord(
id=row["id"],
code=row["code"],
name=row["name"],
code_item_category=row["code_item_category"],
name_item_category=row["name_item_category"],
price=row["price"]
)
for row in rows
]
return ItemSearchResult(
count=len(item_list),
items=item_list
)
except Exception as e:
# raise することで FastMCP が isError=True のエラー応答を自動生成する
# （MCPプロトコル標準の方式）
logger.error(f"get_items エラー: {e}")
raise
# ----------------------------------------------------------------
# get_members ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class MemberRecord(BaseModel):
"""社員マスタの1レコード"""
id: int = Field(description="社員ID（システム内部の一意な数値ID）")
code: str = Field(description="社員コード（業務上の社員識別コード）")
name: str = Field(description="社員名")
class MemberSearchResult(BaseModel):
"""
get_members ツールの戻り値。
count=0 かつ members=[] の場合は検索条件に合致する社員が0件であることを示す（正常応答）。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
count: int = Field(description="取得した社員レコード数。0の場合は検索条件に合致する社員なし。")
members: List[MemberRecord] = Field(
description="社員レコードの一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# get_members ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def get_members(
code_member: Annotated[Optional[str], Field(
description="社員マスタの検索条件。社員コードによる完全一致検索。オプション。",
default=None
)],
name_str_member: Annotated[Optional[str], Field(
description="社員マスタの検索条件。社員名の一部または全部による部分一致検索。オプション。",
default=None
)]
) -> MemberSearchResult:
"""
健康食品の販売システムにおいて、社員マスタの検索を行い、社員データを返す。
引数をすべて省略した場合は全件取得となる。
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
# カーソルはこのtool呼び出しのスコープ内で作成・クローズする
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# DBのFunctionを呼び出す
cur.execute(
"SELECT * FROM public.get_members(%s, %s)",
(code_member, name_str_member)
)
rows = cur.fetchall()
finally:
cur.close()
# RealDictRow → MemberRecord に変換（0件の場合は空リストのまま返す）
member_list = [
MemberRecord(
id=row["id"],
code=row["code"],
name=row["name"]
)
for row in rows
]
return MemberSearchResult(
count=len(member_list),
members=member_list
)
except Exception as e:
# raise することで FastMCP が isError=True のエラー応答を自動生成する
# （MCPプロトコル標準の方式）
logger.error(f"get_members エラー: {e}")
raise
# ----------------------------------------------------------------
# get_sales ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class SalesRecord(BaseModel):
"""販売データの1レコード"""
date_sales_str: str = Field(description="販売日。ISO8601形式の文字列（例: 2024-04-01）")
code_customer: str = Field(description="販売先の顧客のコード")
name_customer: str = Field(description="販売先の顧客の名前")
code_item: str = Field(description="販売された商品のコード")
name_item: str = Field(description="販売された商品の名前")
price_item: int = Field(description="販売された商品の価格（販売単価）")
quantity: int = Field(description="販売された商品の数量")
total_price: int = Field(description="販売価格（合計金額）")
code_item_category: str = Field(description="販売された商品のカテゴリーのコード")
name_item_category: str = Field(description="販売された商品のカテゴリーの名前")
code_member: str = Field(description="販売した社員のコード")
name_member: str = Field(description="販売した社員の名前")
class SalesSearchResult(BaseModel):
"""
get_sales ツールの戻り値。
count=0 かつ sales=[] の場合は検索条件に合致する販売データが0件であることを示す（正常応答）。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
count: int = Field(description="取得した販売レコード数。0の場合は検索条件に合致する販売データなし。")
sales: List[SalesRecord] = Field(
description="販売レコードの一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# get_sales ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def get_sales(
search_period_start: Annotated[str, Field(
description="販売データの検索条件。販売日をひっかける検索期間の開始日。必須。ISO8601形式の年月日（例: 2024-04-01）。"
)],
search_period_end: Annotated[str, Field(
description="販売データの検索条件。販売日をひっかける検索期間の終端日。必須。ISO8601形式の年月日（例: 2024-04-30）。"
)],
code_customer: Annotated[Optional[str], Field(
description="販売データの検索条件。販売先の顧客のコードによる完全一致検索。オプション。",
default=None
)],
name_str_customer: Annotated[Optional[str], Field(
description="販売データの検索条件。販売先の顧客の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_item: Annotated[Optional[str], Field(
description="販売データの検索条件。販売された商品のコードによる完全一致検索。オプション。",
default=None
)],
name_str_item: Annotated[Optional[str], Field(
description="販売データの検索条件。販売された商品の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_item_category: Annotated[Optional[str], Field(
description="販売データの検索条件。販売された商品のカテゴリーのコードによる完全一致検索。オプション。",
default=None
)],
name_str_item_category: Annotated[Optional[str], Field(
description="販売データの検索条件。販売された商品のカテゴリーの名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_member: Annotated[Optional[str], Field(
description="販売データの検索条件。販売した社員のコードによる完全一致検索。オプション。",
default=None
)],
name_str_member: Annotated[Optional[str], Field(
description="販売データの検索条件。販売した社員の名前の一部または全部による部分一致検索。オプション。",
default=None
)]
) -> SalesSearchResult:
"""
健康食品の販売システムにおいて、販売データを検索して返す。集計されておらず、1レコードが1販売に相当する。
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
# カーソルはこのtool呼び出しのスコープ内で作成・クローズする
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# DBのFunctionを呼び出す
cur.execute(
"SELECT * FROM public.get_sales(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(
search_period_start,
search_period_end,
code_customer,
name_str_customer,
code_item,
name_str_item,
code_item_category,
name_str_item_category,
code_member,
name_str_member
)
)
rows = cur.fetchall()
finally:
cur.close()
# RealDictRow → SalesRecord に変換（0件の場合は空リストのまま返す）
# date_sales は psycopg2 が PostgreSQL の date 型を Python の date オブジェクトとして返す
sales_list = [
SalesRecord(
date_sales_str=row["date_sales"].isoformat(),
code_customer=row["code_customer"],
name_customer=row["name_customer"],
code_item=row["code_item"],
name_item=row["name_item"],
price_item=row["price_item"],
quantity=row["quantity"],
total_price=row["total_price"],
code_item_category=row["code_item_category"],
name_item_category=row["name_item_category"],
code_member=row["code_member"],
name_member=row["name_member"]
)
for row in rows
]
return SalesSearchResult(
count=len(sales_list),
sales=sales_list
)
except Exception as e:
# raise することで FastMCP が isError=True のエラー応答を自動生成する
# （MCPプロトコル標準の方式）
logger.error(f"get_sales エラー: {e}")
raise
# ----------------------------------------------------------------
# get_sales_sum_by_customer_item_member ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class SalesSumRecord(BaseModel):
"""顧客毎・商品毎・社員毎の販売集計の1レコード"""
sum_period_start_str: str = Field(description="集計期間の開始日。ISO8601形式の文字列（例: 2024-04-01）")
sum_period_end_str: str = Field(description="集計期間の終端日。ISO8601形式の文字列（例: 2024-04-30）")
code_customer: str = Field(description="販売先の顧客のコード")
name_customer: str = Field(description="販売先の顧客の名前")
code_item: str = Field(description="販売された商品のコード")
name_item: str = Field(description="販売された商品の名前")
quantity: int = Field(description="販売された商品の数量の集計値")
total_price: int = Field(description="販売価格の集計値")
code_item_category: str = Field(description="販売された商品のカテゴリーのコード")
name_item_category: str = Field(description="販売された商品のカテゴリーの名前")
code_member: str = Field(description="販売した社員のコード")
name_member: str = Field(description="販売した社員の名前")
class SalesSumSearchResult(BaseModel):
"""
get_sales_sum_by_customer_item_member ツールの戻り値。
count=0 かつ sales_sum=[] の場合は検索条件に合致する販売データが0件であることを示す（正常応答）。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
count: int = Field(description="取得した集計レコード数。0の場合は検索条件に合致する販売データなし。")
sales_sum: List[SalesSumRecord] = Field(
description="販売集計レコードの一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# get_sales_sum_by_customer_item_member ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def get_sales_sum_by_customer_item_member(
sum_period_start: Annotated[str, Field(
description="集計する販売データの検索条件。販売日をひっかける集計期間の開始日。必須。ISO8601形式の年月日（例: 2024-04-01）。"
)],
sum_period_end: Annotated[str, Field(
description="集計する販売データの検索条件。販売日をひっかける集計期間の終端日。必須。ISO8601形式の年月日（例: 2024-04-30）。"
)],
code_customer: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売先の顧客のコードによる完全一致検索。オプション。",
default=None
)],
name_str_customer: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売先の顧客の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_item: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売された商品のコードによる完全一致検索。オプション。",
default=None
)],
name_str_item: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売された商品の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_item_category: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売された商品のカテゴリーのコードによる完全一致検索。オプション。",
default=None
)],
name_str_item_category: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売された商品のカテゴリーの名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_member: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売した社員のコードによる完全一致検索。オプション。",
default=None
)],
name_str_member: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売した社員の名前の一部または全部による部分一致検索。オプション。",
default=None
)]
) -> SalesSumSearchResult:
"""
指定された集計期間内の販売データを、顧客毎・商品毎・社員毎に集計して返す。
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
# カーソルはこのtool呼び出しのスコープ内で作成・クローズする
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# DBのFunctionを呼び出す
cur.execute(
"SELECT * FROM public.get_sales_sum_by_customer_item_member(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(
sum_period_start,
sum_period_end,
code_customer,
name_str_customer,
code_item,
name_str_item,
code_item_category,
name_str_item_category,
code_member,
name_str_member
)
)
rows = cur.fetchall()
finally:
cur.close()
# RealDictRow → SalesSumRecord に変換（0件の場合は空リストのまま返す）
# sum_period_start / sum_period_end は psycopg2 が PostgreSQL の date 型を
# Python の date オブジェクトとして返す
sales_sum_list = [
SalesSumRecord(
sum_period_start_str=row["sum_period_start"].isoformat(),
sum_period_end_str=row["sum_period_end"].isoformat(),
code_customer=row["code_customer"],
name_customer=row["name_customer"],
code_item=row["code_item"],
name_item=row["name_item"],
quantity=row["quantity"],
total_price=row["total_price"],
code_item_category=row["code_item_category"],
name_item_category=row["name_item_category"],
code_member=row["code_member"],
name_member=row["name_member"]
)
for row in rows
]
return SalesSumSearchResult(
count=len(sales_sum_list),
sales_sum=sales_sum_list
)
except Exception as e:
# raise することで FastMCP が isError=True のエラー応答を自動生成する
# （MCPプロトコル標準の方式）
logger.error(f"get_sales_sum_by_customer_item_member エラー: {e}")
raise
# ----------------------------------------------------------------
# get_sales_sum_by_customer_itemcat_member ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class SalesItemCatSumRecord(BaseModel):
"""顧客毎・商品カテゴリー毎・社員毎の販売集計の1レコード"""
sum_period_start_str: str = Field(description="集計期間の開始日。ISO8601形式の文字列（例: 2024-04-01）")
sum_period_end_str: str = Field(description="集計期間の終端日。ISO8601形式の文字列（例: 2024-04-30）")
code_customer: str = Field(description="販売先の顧客のコード")
name_customer: str = Field(description="販売先の顧客の名前")
total_price: int = Field(description="販売価格の集計値")
code_item_category: str = Field(description="販売された商品のカテゴリーのコード")
name_item_category: str = Field(description="販売された商品のカテゴリーの名前")
code_member: str = Field(description="販売した社員のコード")
name_member: str = Field(description="販売した社員の名前")
class SalesItemCatSumSearchResult(BaseModel):
"""
get_sales_sum_by_customer_itemcat_member ツールの戻り値。
count=0 かつ sales_sum=[] の場合は検索条件に合致する販売データが0件であることを示す（正常応答）。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
count: int = Field(description="取得した集計レコード数。0の場合は検索条件に合致する販売データなし。")
sales_sum: List[SalesItemCatSumRecord] = Field(
description="販売集計レコードの一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# get_sales_sum_by_customer_itemcat_member ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def get_sales_sum_by_customer_itemcat_member(
sum_period_start: Annotated[str, Field(
description="集計する販売データの検索条件。販売日をひっかける集計期間の開始日。必須。ISO8601形式の年月日（例: 2024-04-01）。"
)],
sum_period_end: Annotated[str, Field(
description="集計する販売データの検索条件。販売日をひっかける集計期間の終端日。必須。ISO8601形式の年月日（例: 2024-04-30）。"
)],
code_customer: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売先の顧客のコードによる完全一致検索。オプション。",
default=None
)],
name_str_customer: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売先の顧客の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_item: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売された商品のコードによる完全一致検索。オプション。",
default=None
)],
name_str_item: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売された商品の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_item_category: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売された商品のカテゴリーのコードによる完全一致検索。オプション。",
default=None
)],
name_str_item_category: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売された商品のカテゴリーの名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_member: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売した社員のコードによる完全一致検索。オプション。",
default=None
)],
name_str_member: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売した社員の名前の一部または全部による部分一致検索。オプション。",
default=None
)]
) -> SalesItemCatSumSearchResult:
"""
指定された集計期間内の販売データを、顧客毎・商品カテゴリー毎・社員毎に集計して返す。
"""
try:
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# DBのFunctionを呼び出す
cur.execute(
"SELECT * FROM public.get_sales_sum_by_customer_itemcat_member(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(
sum_period_start,
sum_period_end,
code_customer,
name_str_customer,
code_item,
name_str_item,
code_item_category,
name_str_item_category,
code_member,
name_str_member
)
)
rows = cur.fetchall()
finally:
cur.close()
# RealDictRow → SalesItemCatSumRecord に変換（0件の場合は空リストのまま返す）
# sum_period_start / sum_period_end は psycopg2 が PostgreSQL の date 型を
# Python の date オブジェクトとして返す
sales_sum_list = [
SalesItemCatSumRecord(
sum_period_start_str=row["sum_period_start"].isoformat(),
sum_period_end_str=row["sum_period_end"].isoformat(),
code_customer=row["code_customer"],
name_customer=row["name_customer"],
total_price=row["total_price"],
code_item_category=row["code_item_category"],
name_item_category=row["name_item_category"],
code_member=row["code_member"],
name_member=row["name_member"]
)
for row in rows
]
return SalesItemCatSumSearchResult(
count=len(sales_sum_list),
sales_sum=sales_sum_list
)
except Exception as e:
logger.error(f"get_sales_sum_by_customer_itemcat_member エラー: {e}")
raise
# ----------------------------------------------------------------
# get_sales_sum_by_customer ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class SalesSumByCustomerRecord(BaseModel):
"""顧客毎の販売集計の1レコード"""
sum_period_start_str: str = Field(description="集計期間の開始日。ISO8601形式の文字列（例: 2024-04-01）")
sum_period_end_str: str = Field(description="集計期間の終端日。ISO8601形式の文字列（例: 2024-04-30）")
code_customer: str = Field(description="販売先の顧客のコード")
name_customer: str = Field(description="販売先の顧客の名前")
total_price: int = Field(description="その顧客の、集計期間での販売価格（合計金額）の集計値")
avg_total_price_all_customer: float = Field(description="集計期間での、全顧客間の平均売上合計額")
class SalesSumByCustomerSearchResult(BaseModel):
"""
get_sales_sum_by_customer ツールの戻り値。
count=0 かつ sales_sum=[] の場合は検索条件に合致する顧客が0件であることを示す（正常応答）。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
count: int = Field(description="取得した集計レコード数。0の場合は検索条件に合致する顧客なし。")
sales_sum: List[SalesSumByCustomerRecord] = Field(
description="顧客毎の販売集計レコードの一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# get_sales_sum_by_customer ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def get_sales_sum_by_customer(
sum_period_start: Annotated[str, Field(
description="集計する販売データの検索条件。販売日をひっかける集計期間の開始日。必須。ISO8601形式の年月日（例: 2024-04-01）。"
)],
sum_period_end: Annotated[str, Field(
description="集計する販売データの検索条件。販売日をひっかける集計期間の終端日。必須。ISO8601形式の年月日（例: 2024-04-30）。"
)],
code_customer: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売先の顧客のコードによる完全一致検索。オプション。",
default=None
)],
name_str_customer: Annotated[Optional[str], Field(
description="集計する販売データの検索条件。販売先の顧客の名前の一部または全部による部分一致検索。オプション。",
default=None
)]
) -> SalesSumByCustomerSearchResult:
"""
指定された集計期間での顧客毎の販売額合計と、全顧客の販売額合計の平均値を返す。
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
# カーソルはこのtool呼び出しのスコープ内で作成・クローズする
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# DBのFunctionを呼び出す
cur.execute(
"SELECT * FROM public.get_sales_sum_by_customer(%s, %s, %s, %s)",
(
sum_period_start,
sum_period_end,
code_customer,
name_str_customer
)
)
rows = cur.fetchall()
finally:
cur.close()
# RealDictRow → SalesSumByCustomerRecord に変換（0件の場合は空リストのまま返す）
# sum_period_start / sum_period_end は psycopg2 が PostgreSQL の date 型を
# Python の date オブジェクトとして返す
sales_sum_list = [
SalesSumByCustomerRecord(
sum_period_start_str=row["sum_period_start"].isoformat(),
sum_period_end_str=row["sum_period_end"].isoformat(),
code_customer=row["code_customer"],
name_customer=row["name_customer"],
total_price=row["total_price"],
avg_total_price_all_customer=row["avg_total_price_all_customer"]
)
for row in rows
]
return SalesSumByCustomerSearchResult(
count=len(sales_sum_list),
sales_sum=sales_sum_list
)
except Exception as e:
# raise することで FastMCP が isError=True のエラー応答を自動生成する
# （MCPプロトコル標準の方式）
logger.error(f"get_sales_sum_by_customer エラー: {e}")
raise
if __name__ == "__main__":
mcp.run(
transport="streamable-http",
host="192.168.x.x", # このMCPサーバーを実行するマシンのプライベートIPアドレス
port=8001
)
FastMCPは、「@tool」が付加された関数をMCPツールと判断し、その中身を見て、MCPサーバーの全ツールのSchemaを自動生成します。
inputSchemaだけでなく、実のあるoutputSchemaを生成させるために、各MCPツールの戻り値をBaseModelの派生クラスで詳細に定義しました。
MCPサーバー「inquiry_data_mng」
本稿の「MCPサーバー2」です。
問い合わせ関連データ（注文・問い合わせ・問い合わせ対応データ）の取得や更新をします。
FastMCPを使用しています。
クリックして表示
| 項目 | 内容 |
|---|---|
| サーバー名 | inquiry_data_mng |
| ファイル名 | mcp_server_inquiry_data_mng.py |
| Transport | Streamable HTTP |
| 認証 | 静的Bearer Token |
MCPサーバー「sales_stat」と同じで、認証に静的Bearer Tokenを用いています。
トークンが平文でHTTPリクエストに含まれるため、本来はHTTPSで通信すべきですが、閉じられた宅内LANで使用しているだけなので、そこまではしませんでした。
ツール
各ツール内部で、適したDBのFunctionを使用しています。
| 名称 | 機能 |
|---|---|
| get_orders | DBから注文データを検索して返す。集計無し。販売データ1個＝戻り値1行。 |
| get_inquiries | DBから問い合わせデータを取得して返す。 戻り値1行は、 「1件の問い合わせ」＋「その問い合わせの対象である1件の注文」 という構成である。 どんな注文に対するどんな問い合わせなのか、戻り値の1行を見ればわかる。 |
| get_inquiry_responses | DBから問い合わせ対応データを取得して返す。 戻り値1行は、 「1件の問い合わせ対応」＋「その対応をした1件の問い合わせ」＋「その問い合わせの対象である1件の注文」 という構成である。 どんな注文に対するどんな問い合わせで、それに対してどんな対応をしたのか、戻り値の1行を見ればわかる。 |
| update_inquiry | DBの問い合わせデータ1件を更新する。 限られたフィールドのみ更新可。 |
コード
MCPサーバー「mcp_server_inquiry_data_mng.py」
from fastmcp import FastMCP
import os
from typing import Annotated, List
from pydantic import BaseModel, Field
from typing import Optional
import logging
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import AsyncIterator
import psycopg2
import psycopg2.extras
from pathlib import Path
from dotenv import load_dotenv
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier
load_dotenv(Path(__file__).resolve().parent / ".env")
# ロギング設定（出力フォーマット追加）
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [INQUIRY-DATA-MNG-SERVER] %(message)s"
)
logger = logging.getLogger("fast-mcp-server-inquiry-mng")
# DB接続をグローバル変数として保持
# ※カーソルはtool呼び出しごとに新規作成する
db_connection = None
@dataclass
class AppContext:
server_name: str = "MCP Server for inquiry data management"
start_time: str = ""
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
global db_connection
try:
yield AppContext()
finally:
# 終了時：クリーンアップ
if db_connection is not None:
db_connection.close()
db_connection = None
# verifierを作り、FastMCPに渡す
_bearer_token = os.environ.get("MCP_INQUIRY_DATA_MNG_BEARER_TOKEN")
if not _bearer_token:
raise ValueError("環境変数 'MCP_INQUIRY_DATA_MNG_BEARER_TOKEN' が設定されていません。")
verifier = StaticTokenVerifier(
tokens={
_bearer_token: {
"client_id": "inquiry-data-mng-client",
"scopes": ["data:access"]
}
},
required_scopes=["data:access"]
)
mcp = FastMCP("Inquiry_data_management_server",
instructions="健康食品の販売システムにおいて、問い合わせのデータ管理を行う。注文データ、問い合わせデータ、問い合わせ対応データの取得を行う。問い合わせデータの更新を行う。",
lifespan=app_lifespan,
auth=verifier)
def get_db_connection():
"""
DB接続を返す。接続が無い or 切断済みの場合のみ新規作成。
カーソルは呼び出し元で都度作成すること。
"""
global db_connection
if db_connection is None or db_connection.closed:
try:
db_dsn = os.environ.get('DBDSN')
if not db_dsn:
raise ValueError("環境変数 'DBDSN' が設定されていません。")
# DSN (Data Source Name) を使用して接続
db_connection = psycopg2.connect(dsn=db_dsn, sslmode='require')
# autocommit モードを設定
db_connection.autocommit = True
except Exception as e:
raise
return db_connection
# ----------------------------------------------------------------
# get_orders ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class OrderRecord(BaseModel):
"""注文データの1レコード"""
id: int = Field(description="注文ID（システム内部の一意な数値ID）")
code_order: str = Field(description="注文番号")
date_order_str: str = Field(description="注文日。ISO8601形式の文字列（例: 2024-04-01）")
code_customer: str = Field(description="注文した顧客のコード")
name_customer: str = Field(description="注文した顧客の名前")
code_member: str = Field(description="注文を受けた社員のコード")
name_member: str = Field(description="注文を受けた社員の名前")
code_item: str = Field(description="注文された商品のコード")
name_item: str = Field(description="注文された商品の名前")
quantity: int = Field(description="注文された商品の数量")
total_price: int = Field(description="注文金額。合計金額。")
date_delivery_yotei_str: str = Field(description="納品予定日。ISO8601形式の文字列（例: 2024-04-15）")
date_delivery_str: Optional[str] = Field(description="納品日。ISO8601形式の文字列（例: 2024-04-20）。未納品の場合はnull。")
is_delivered: bool = Field(description="納品済みか。trueなら納品済み、falseなら未納品。")
date_shipped_str: Optional[str] = Field(description="発送日。ISO8601形式の文字列（例: 2024-04-18）。未発送の場合はnull。")
is_shipped: bool = Field(description="発送済みか。trueなら発送済み、falseなら未発送。")
code_item_category: str = Field(description="注文された商品のカテゴリーのコード")
name_item_category: str = Field(description="注文された商品のカテゴリーの名前")
is_delivered_on_time: Optional[bool] = Field(description="納期に間に合って納品されたか。trueなら間に合った、falseなら遅延。未納品の場合はnull。")
class OrderSearchResult(BaseModel):
"""
get_orders ツールの戻り値。
count=0 かつ orders=[] の場合は検索条件に合致する注文が0件であることを示す（正常応答）。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
count: int = Field(description="取得した注文レコード数。0の場合は検索条件に合致する注文なし。")
orders: List[OrderRecord] = Field(
description="注文レコードの一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# get_orders ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def get_orders(
date_order_begin: Annotated[str, Field(
description="注文データの検索条件。注文日をひっかける検索期間の開始日。必須。ISO8601形式の年月日（例: 2024-04-01）。"
)],
date_order_end: Annotated[str, Field(
description="注文データの検索条件。注文日をひっかける検索期間の終了日。必須。ISO8601形式の年月日（例: 2024-04-30）。"
)],
code_order: Annotated[Optional[str], Field(
description="注文データの検索条件。注文番号による完全一致検索。オプション。",
default=None
)],
code_customer: Annotated[Optional[str], Field(
description="注文データの検索条件。注文した顧客のコードによる完全一致検索。オプション。",
default=None
)],
name_str_customer: Annotated[Optional[str], Field(
description="注文データの検索条件。注文した顧客の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_item: Annotated[Optional[str], Field(
description="注文データの検索条件。注文された商品のコードによる完全一致検索。オプション。",
default=None
)],
name_str_item: Annotated[Optional[str], Field(
description="注文された商品の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_item_category: Annotated[Optional[str], Field(
description="注文データの検索条件。注文された商品のカテゴリーのコードによる完全一致検索。オプション。",
default=None
)],
name_str_item_category: Annotated[Optional[str], Field(
description="注文された商品のカテゴリーの名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_member: Annotated[Optional[str], Field(
description="注文データの検索条件。注文を受けた社員のコードによる完全一致検索。オプション。",
default=None
)],
name_str_member: Annotated[Optional[str], Field(
description="注文を受けた社員の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
date_delivery_yotei_begin: Annotated[Optional[str], Field(
description="注文データの検索条件。納品予定日をひっかける検索期間の開始日。オプション。ISO8601形式の年月日（例: 2024-04-01）。",
default=None
)],
date_delivery_yotei_end: Annotated[Optional[str], Field(
description="注文データの検索条件。納品予定日をひっかける検索期間の終了日。オプション。ISO8601形式の年月日（例: 2024-04-30）。",
default=None
)],
date_delivery_begin: Annotated[Optional[str], Field(
description="注文データの検索条件。納品日をひっかける検索期間の開始日。オプション。ISO8601形式の年月日（例: 2024-04-01）。",
default=None
)],
date_delivery_end: Annotated[Optional[str], Field(
description="注文データの検索条件。納品日をひっかける検索期間の終了日。オプション。ISO8601形式の年月日（例: 2024-04-30）。",
default=None
)],
is_delivered: Annotated[Optional[bool], Field(
description="注文データの検索条件。納品済みか。trueなら納品済みのみ、falseなら未納品のみ。オプション。",
default=None
)],
date_shipped_begin: Annotated[Optional[str], Field(
description="注文データの検索条件。発送日をひっかける検索期間の開始日。オプション。ISO8601形式の年月日（例: 2024-04-01）。",
default=None
)],
date_shipped_end: Annotated[Optional[str], Field(
description="注文データの検索条件。発送日をひっかける検索期間の終了日。オプション。ISO8601形式の年月日（例: 2024-04-30）。",
default=None
)],
is_shipped: Annotated[Optional[bool], Field(
description="注文データの検索条件。発送済みか。trueなら発送済みのみ、falseなら未発送のみ。オプション。",
default=None
)],
is_delivered_on_time: Annotated[Optional[bool], Field(
description="注文データの検索条件。納期に間に合って納品されたか。trueなら間に合った注文のみ、falseなら遅延した注文のみ。オプション。",
default=None
)]
) -> OrderSearchResult:
"""
健康食品の販売システムにおいて、注文データを検索して返す。1レコードが1注文に相当する。
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
# カーソルはこのtool呼び出しのスコープ内で作成・クローズする
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# DBのFunctionを呼び出す
cur.execute(
"SELECT * FROM public.get_orders(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(
date_order_begin,
date_order_end,
code_order,
code_customer,
name_str_customer,
code_item,
name_str_item,
code_item_category,
name_str_item_category,
code_member,
name_str_member,
date_delivery_yotei_begin,
date_delivery_yotei_end,
date_delivery_begin,
date_delivery_end,
is_delivered,
date_shipped_begin,
date_shipped_end,
is_shipped,
is_delivered_on_time
)
)
rows = cur.fetchall()
finally:
cur.close()
# RealDictRow → OrderRecord に変換（0件の場合は空リストのまま返す）
# date系カラムは psycopg2 が PostgreSQL の date 型を Python の date オブジェクトとして返す
# nullable な date カラム（date_delivery, date_shipped）は None の場合がある
order_list = [
OrderRecord(
id=row["id"],
code_order=row["code_order"],
date_order_str=row["date_order"].isoformat(),
code_customer=row["code_customer"],
name_customer=row["name_customer"],
code_member=row["code_member"],
name_member=row["name_member"],
code_item=row["code_item"],
name_item=row["name_item"],
quantity=row["quantity"],
total_price=row["total_price"],
date_delivery_yotei_str=row["date_delivery_yotei"].isoformat(),
date_delivery_str=row["date_delivery"].isoformat() if row["date_delivery"] else None,
is_delivered=row["is_delivered"],
date_shipped_str=row["date_shipped"].isoformat() if row["date_shipped"] else None,
is_shipped=row["is_shipped"],
code_item_category=row["code_item_category"],
name_item_category=row["name_item_category"],
is_delivered_on_time=row["is_delivered_on_time"]
)
for row in rows
]
return OrderSearchResult(
count=len(order_list),
orders=order_list
)
except Exception as e:
# raise することで FastMCP が isError=True のエラー応答を自動生成する
# （MCPプロトコル標準の方式）
logger.error(f"get_orders エラー: {e}")
raise
# ----------------------------------------------------------------
# get_inquiries ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class InquiryRecord(BaseModel):
"""問い合わせデータの1レコード。「1件の問い合わせ」＋「その問い合わせの対象である1件の注文」という構成。"""
id: int = Field(description="問い合わせID（システム内部の一意な数値ID）")
code_inquiry: str = Field(description="問い合わせ番号")
date_inquiry_str: str = Field(description="問い合わせ日。ISO8601形式の文字列（例: 2024-04-01）")
code_customer: str = Field(description="問い合わせした顧客のコード")
name_customer: str = Field(description="問い合わせした顧客の名前")
id_order: int = Field(description="問い合わせの対象である注文の注文ID")
code_order: str = Field(description="問い合わせの対象である注文の注文番号")
is_begun: bool = Field(description="問い合わせ対応に着手済みか。trueなら着手済み、falseなら未着手。")
is_ended: bool = Field(description="問い合わせ対応が終了済みか。trueなら終了、falseなら未終了。")
body_inquiry: str = Field(description="問い合わせ本文")
sug_num_res_route: Optional[int] = Field(description="AI提案の対応ルート番号。1：自動送信メールで回答。2：人間が対応。未設定の場合はnull。")
sug_name_res_route: Optional[str] = Field(description="AI提案の対応ルート名。未設定の場合はnull。")
summary_inquiry: Optional[str] = Field(description="問い合わせ本文の要約。未設定の場合はnull。")
hint_res: Optional[str] = Field(description="対応のヒント。未設定の場合はnull。")
draft_auto_res_message: Optional[str] = Field(description="自動返信メールの文案。未設定の場合はnull。")
is_urgent: Optional[bool] = Field(description="緊急性の有無。trueなら緊急、falseなら緊急でない。nullの場合は未設定。")
is_urgent_internal_notified: Optional[bool] = Field(description="緊急性有りの場合、社内通知したか。社内通知済みならtrue。nullの場合は未通知か未設定。")
date_order_str: str = Field(description="問い合わせの対象である注文の注文日。ISO8601形式の文字列（例: 2024-04-01）")
code_customer_order: str = Field(description="問い合わせの対象である注文の顧客コード")
name_customer_order: str = Field(description="問い合わせの対象である注文の顧客名")
code_member_order: str = Field(description="問い合わせの対象である注文を受けた社員コード")
name_member_order: str = Field(description="問い合わせの対象である注文を受けた社員名")
code_item_order: str = Field(description="問い合わせの対象である注文の商品コード")
name_item_order: str = Field(description="問い合わせの対象である注文の商品名")
quantity_order: int = Field(description="問い合わせの対象である注文の数量")
total_price_order: int = Field(description="問い合わせの対象である注文の合計金額")
date_delivery_yotei_order_str: str = Field(description="問い合わせの対象である注文の納品予定日。ISO8601形式の文字列（例: 2024-04-15）")
date_delivery_order_str: Optional[str] = Field(description="問い合わせの対象である注文の納品日。ISO8601形式の文字列（例: 2024-04-20）。未納品の場合はnull。")
is_delivered_order: bool = Field(description="問い合わせの対象である注文が納品済みか。trueなら納品済み、falseなら未納品。")
date_shipped_order_str: Optional[str] = Field(description="問い合わせの対象である注文の発送日。ISO8601形式の文字列（例: 2024-04-18）。未発送の場合はnull。")
is_shipped_order: bool = Field(description="問い合わせの対象である注文が発送済みか。trueなら発送済み、falseなら未発送。")
code_item_category_order: str = Field(description="問い合わせの対象である注文の商品カテゴリーコード")
name_item_category_order: str = Field(description="問い合わせの対象である注文の商品カテゴリー名")
is_delivered_on_time_order: Optional[bool] = Field(description="問い合わせの対象である注文が納期に間に合ったか。trueなら間に合った、falseなら遅延。未納品の場合はnull。")
class InquirySearchResult(BaseModel):
"""
get_inquiries ツールの戻り値。
count=0 かつ inquiries=[] の場合は検索条件に合致する問い合わせが0件であることを示す（正常応答）。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
count: int = Field(description="取得した問い合わせレコード数。0の場合は検索条件に合致する問い合わせなし。")
inquiries: List[InquiryRecord] = Field(
description="問い合わせレコードの一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# get_inquiries ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def get_inquiries(
date_inquiry_begin: Annotated[str, Field(
description="問い合わせデータの検索条件。問い合わせ日をひっかける検索期間の開始日。必須。ISO8601形式の年月日（例: 2024-04-01）。"
)],
date_inquiry_end: Annotated[str, Field(
description="問い合わせデータの検索条件。問い合わせ日をひっかける検索期間の終了日。必須。ISO8601形式の年月日（例: 2024-04-30）。"
)],
code_inquiry: Annotated[Optional[str], Field(
description="問い合わせデータの検索条件。問い合わせ番号による完全一致検索。オプション。",
default=None
)],
code_customer: Annotated[Optional[str], Field(
description="問い合わせデータの検索条件。問い合わせした顧客のコードによる完全一致検索。オプション。",
default=None
)],
name_str_customer: Annotated[Optional[str], Field(
description="問い合わせデータの検索条件。問い合わせした顧客の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_order: Annotated[Optional[str], Field(
description="問い合わせデータの検索条件。問い合わせの対象である注文の注文番号による完全一致検索。オプション。",
default=None
)],
date_order_begin: Annotated[Optional[str], Field(
description="問い合わせデータの検索条件。問い合わせの対象である注文の注文日をひっかける検索期間の開始日。オプション。ISO8601形式の年月日（例: 2024-04-01）。",
default=None
)],
date_order_end: Annotated[Optional[str], Field(
description="問い合わせデータの検索条件。問い合わせの対象である注文の注文日をひっかける検索期間の終了日。オプション。ISO8601形式の年月日（例: 2024-04-30）。",
default=None
)],
code_item_order: Annotated[Optional[str], Field(
description="問い合わせデータの検索条件。問い合わせの対象である注文の商品コードによる完全一致検索。オプション。",
default=None
)],
name_str_item_order: Annotated[Optional[str], Field(
description="問い合わせデータの検索条件。問い合わせの対象である注文の商品名の一部または全部による部分一致検索。オプション。",
default=None
)],
is_begun: Annotated[Optional[bool], Field(
description="問い合わせデータの検索条件。問い合わせ対応に着手済みか。trueなら着手済みのみ、falseなら未着手のみ。オプション。",
default=None
)],
is_ended: Annotated[Optional[bool], Field(
description="問い合わせデータの検索条件。問い合わせ対応は終了しているか。trueなら終了のみ、falseなら未終了のみ。オプション。",
default=None
)],
is_urgent: Annotated[Optional[bool], Field(
description="問い合わせデータの検索条件。緊急か。trueなら緊急のみ、falseなら緊急でないもののみ。オプション。",
default=None
)],
sug_num_res_route: Annotated[Optional[int], Field(
description="問い合わせデータの検索条件。AI提案の対応ルート番号による完全一致検索。1：「自動送信メールで回答」提案のみ。2：「人間が対応」提案のみ。オプション。",
default=None
)]
) -> InquirySearchResult:
"""
健康食品の販売システムにおいて、問い合わせデータを検索して返す。1レコードが1件の問い合わせに相当する。
その1レコードは、「1件の問い合わせ」＋「その問い合わせの対象である1件の注文」という構成である。
どんな注文に対するどんな問い合わせなのか、戻り値の1レコードを見ればわかる。
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
# カーソルはこのtool呼び出しのスコープ内で作成・クローズする
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# DBのFunctionを呼び出す
cur.execute(
"SELECT * FROM public.get_inquiries(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(
date_inquiry_begin,
date_inquiry_end,
code_inquiry,
code_customer,
name_str_customer,
code_order,
date_order_begin,
date_order_end,
code_item_order,
name_str_item_order,
is_begun,
is_ended,
is_urgent,
sug_num_res_route
)
)
rows = cur.fetchall()
finally:
cur.close()
# RealDictRow → InquiryRecord に変換（0件の場合は空リストのまま返す）
# date系カラムは psycopg2 が PostgreSQL の date 型を Python の date オブジェクトとして返す
# nullable な date カラムは None の場合がある
inquiry_list = [
InquiryRecord(
id=row["id"],
code_inquiry=row["code_inquiry"],
date_inquiry_str=row["date_inquiry"].isoformat(),
code_customer=row["code_customer"],
name_customer=row["name_customer"],
id_order=row["id_order"],
code_order=row["code_order"],
is_begun=row["is_begun"],
is_ended=row["is_ended"],
body_inquiry=row["body_inquiry"],
sug_num_res_route=row["sug_num_res_route"],
sug_name_res_route=row["sug_name_res_route"],
summary_inquiry=row["summary_inquiry"],
hint_res=row["hint_res"],
draft_auto_res_message=row["draft_auto_res_message"],
is_urgent=row["is_urgent"],
is_urgent_internal_notified=row["is_urgent_internal_notified"],
date_order_str=row["date_order"].isoformat(),
code_customer_order=row["code_customer_order"],
name_customer_order=row["name_customer_order"],
code_member_order=row["code_member_order"],
name_member_order=row["name_member_order"],
code_item_order=row["code_item_order"],
name_item_order=row["name_item_order"],
quantity_order=row["quantity_order"],
total_price_order=row["total_price_order"],
date_delivery_yotei_order_str=row["date_delivery_yotei_order"].isoformat(),
date_delivery_order_str=row["date_delivery_order"].isoformat() if row["date_delivery_order"] else None,
is_delivered_order=row["is_delivered_order"],
date_shipped_order_str=row["date_shipped_order"].isoformat() if row["date_shipped_order"] else None,
is_shipped_order=row["is_shipped_order"],
code_item_category_order=row["code_item_category_order"],
name_item_category_order=row["name_item_category_order"],
is_delivered_on_time_order=row["is_delivered_on_time_order"]
)
for row in rows
]
return InquirySearchResult(
count=len(inquiry_list),
inquiries=inquiry_list
)
except Exception as e:
# raise することで FastMCP が isError=True のエラー応答を自動生成する
# （MCPプロトコル標準の方式）
logger.error(f"get_inquiries エラー: {e}")
raise
# ----------------------------------------------------------------
# get_inquiry_responses ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class InquiryResponseRecord(BaseModel):
"""問い合わせ対応データの1レコード。「1件の問い合わせ対応」＋「対応した1件の問い合わせ」＋「その問い合わせの対象である1件の注文」という構成。"""
# -- 問い合わせ対応 --
id: int = Field(description="問い合わせ対応ID（システム内部の一意な数値ID）")
code_inquiry_res: str = Field(description="問い合わせ対応番号")
date_inquiry_res_str: str = Field(description="問い合わせ対応日。ISO8601形式の文字列（例: 2024-04-10）")
id_inquiry: int = Field(description="問い合わせID")
code_inquiry: str = Field(description="問い合わせ番号")
num_res_route: int = Field(description="実際の対応ルート番号。1：自動送信メールで回答。2：人間が対応。")
name_res_route: str = Field(description="実際の対応ルート名")
summary_res: Optional[str] = Field(description="対応内容の概要。未設定の場合はnull。")
auto_res_message: Optional[str] = Field(description="自動送信メールで回答した本文。人間が対応した場合や未設定の場合はnull。")
# -- 問い合わせ --
date_inquiry_inquiry_str: str = Field(description="問い合わせ日。ISO8601形式の文字列（例: 2024-04-01）")
code_customer_inquiry: str = Field(description="問い合わせをした顧客コード")
name_customer_inquiry: str = Field(description="問い合わせをした顧客名")
body_inquiry: str = Field(description="問い合わせ本文")
sug_num_res_route_inquiry: int = Field(description="問い合わせに対するAI提案の対応ルートの番号。1：自動送信メールで回答。2：人間が対応。")
sug_name_res_route_inquiry: str = Field(description="問い合わせに対するAI提案の対応ルート名")
summary_inquiry: Optional[str] = Field(description="問い合わせ本文の要約。未設定の場合はnull。")
hint_res_inquiry: Optional[str] = Field(description="問い合わせに対する対応ヒント。未設定の場合はnull。")
draft_auto_res_message_inquiry: Optional[str] = Field(description="問い合わせに対する自動送信メール文案。未設定の場合はnull。")
is_urgent_inquiry: Optional[bool] = Field(description="問い合わせの緊急性の有無。trueなら緊急、falseなら緊急でない。nullの場合は未判定。")
# -- 注文 --
id_order: int = Field(description="問い合わせの対象である注文ID")
code_order: str = Field(description="問い合わせの対象である注文番号")
date_order_order_str: str = Field(description="問い合わせの対象である注文の注文日。ISO8601形式の文字列（例: 2024-03-15）")
code_customer_order: str = Field(description="問い合わせの対象である注文の顧客コード")
name_customer_order: str = Field(description="問い合わせの対象である注文の顧客名")
code_member_order: str = Field(description="問い合わせの対象である注文を受けた社員コード")
name_member_order: str = Field(description="問い合わせの対象である注文を受けた社員名")
code_item_order: str = Field(description="問い合わせの対象である注文の商品コード")
name_item_order: str = Field(description="問い合わせの対象である注文の商品名")
quantity_order: int = Field(description="問い合わせの対象である注文の数量")
total_price_order: int = Field(description="問い合わせの対象である注文の合計金額")
date_delivery_yotei_order_str: str = Field(description="問い合わせの対象である注文の納品予定日。ISO8601形式の文字列（例: 2024-04-15）")
date_delivery_order_str: Optional[str] = Field(description="問い合わせの対象である注文の納品日。ISO8601形式の文字列（例: 2024-04-20）。未納品の場合はnull。")
is_delivered_order: bool = Field(description="問い合わせの対象である注文が納品済みか。trueなら納品済み、falseなら未納品。")
date_shipped_order_str: Optional[str] = Field(description="問い合わせの対象である注文の発送日。ISO8601形式の文字列（例: 2024-04-18）。未発送の場合はnull。")
is_shipped_order: bool = Field(description="問い合わせの対象である注文が発送済みか。trueなら発送済み、falseなら未発送。")
code_item_category_order: str = Field(description="問い合わせの対象である注文の商品カテゴリーコード")
name_item_category_order: str = Field(description="問い合わせの対象である注文の商品カテゴリー名")
is_delivered_on_time_order: Optional[bool] = Field(description="問い合わせの対象である注文が納期に間に合ったか。trueなら間に合った、falseなら遅延。未納品の場合はnull。")
class InquiryResponseSearchResult(BaseModel):
"""
get_inquiry_responses ツールの戻り値。
count=0 かつ inquiry_responses=[] の場合は検索条件に合致する問い合わせ対応が0件であることを示す（正常応答）。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
count: int = Field(description="取得した問い合わせ対応レコード数。0の場合は検索条件に合致する問い合わせ対応なし。")
inquiry_responses: List[InquiryResponseRecord] = Field(
description="問い合わせ対応レコードの一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# get_inquiry_responses ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def get_inquiry_responses(
date_inquiry_res_begin: Annotated[str, Field(
description="問い合わせ対応データの検索条件。問い合わせ対応日をひっかける検索期間の開始日。必須。ISO8601形式の年月日（例: 2024-04-01）。"
)],
date_inquiry_res_end: Annotated[str, Field(
description="問い合わせ対応データの検索条件。問い合わせ対応日をひっかける検索期間の終了日。必須。ISO8601形式の年月日（例: 2024-04-30）。"
)],
code_inquiry_res: Annotated[Optional[str], Field(
description="問い合わせ対応データの検索条件。問い合わせ対応番号による完全一致検索。オプション。",
default=None
)],
num_res_route: Annotated[Optional[int], Field(
description="問い合わせ対応データの検索条件。実際の対応ルート番号による完全一致検索。1：自動送信メールで回答。2：人間が対応。オプション。",
default=None
)],
code_inquiry: Annotated[Optional[str], Field(
description="問い合わせ対応データの検索条件。問い合わせ番号による完全一致検索。オプション。",
default=None
)],
date_inquiry_begin: Annotated[Optional[str], Field(
description="問い合わせ対応データの検索条件。問い合わせ日をひっかける検索期間の開始日。オプション。ISO8601形式の年月日（例: 2024-04-01）。",
default=None
)],
date_inquiry_end: Annotated[Optional[str], Field(
description="問い合わせ対応データの検索条件。問い合わせ日をひっかける検索期間の終了日。オプション。ISO8601形式の年月日（例: 2024-04-30）。",
default=None
)],
code_customer: Annotated[Optional[str], Field(
description="問い合わせ対応データの検索条件。問い合わせした顧客のコードによる完全一致検索。オプション。",
default=None
)],
name_str_customer: Annotated[Optional[str], Field(
description="問い合わせ対応データの検索条件。問い合わせした顧客の名前の一部または全部による部分一致検索。オプション。",
default=None
)],
code_item_order: Annotated[Optional[str], Field(
description="問い合わせ対応データの検索条件。問い合わせの対象である注文の商品コードによる完全一致検索。オプション。",
default=None
)],
name_str_item_order: Annotated[Optional[str], Field(
description="問い合わせ対応データの検索条件。問い合わせの対象である注文の商品名の一部または全部による部分一致検索。オプション。",
default=None
)],
is_urgent: Annotated[Optional[bool], Field(
description="問い合わせ対応データの検索条件。問い合わせの緊急性の有無。trueなら緊急性有りのみ、falseなら緊急性無しのみ。オプション。",
default=None
)],
sug_num_res_route: Annotated[Optional[int], Field(
description="問い合わせ対応データの検索条件。AI提案の対応ルート番号による完全一致検索。1：自動送信メールで回答。2：人間が対応。オプション。",
default=None
)]
) -> InquiryResponseSearchResult:
"""
健康食品の販売システムにおいて、問い合わせ対応データを検索して返す。1レコードが1件の問い合わせ対応に相当する。
その1レコードは、「1件の問い合わせ対応」＋「対応した1件の問い合わせ」＋「その問い合わせの対象である1件の注文」という構成である。
どんな注文に対するどんな問い合わせで、それに対してどんな対応をしたのか、戻り値の1レコードを見ればわかる。
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
# カーソルはこのtool呼び出しのスコープ内で作成・クローズする
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
# DBのFunctionを呼び出す
cur.execute(
"SELECT * FROM public.get_inquiry_responses(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(
date_inquiry_res_begin,
date_inquiry_res_end,
code_inquiry_res,
num_res_route,
code_inquiry,
date_inquiry_begin,
date_inquiry_end,
code_customer,
name_str_customer,
code_item_order,
name_str_item_order,
is_urgent,
sug_num_res_route
)
)
rows = cur.fetchall()
finally:
cur.close()
# RealDictRow → InquiryResponseRecord に変換（0件の場合は空リストのまま返す）
# date系カラムは psycopg2 が PostgreSQL の date 型を Python の date オブジェクトとして返す
# nullable な date カラム（date_delivery_order, date_shipped_order）は None の場合がある
inquiry_response_list = [
InquiryResponseRecord(
# -- 問い合わせ対応 --
id=row["id"],
code_inquiry_res=row["code_inquiry_res"],
date_inquiry_res_str=row["date_inquiry_res"].isoformat(),
id_inquiry=row["id_inquiry"],
code_inquiry=row["code_inquiry"],
num_res_route=row["num_res_route"],
name_res_route=row["name_res_route"],
summary_res=row["summary_res"],
auto_res_message=row["auto_res_message"],
# -- 問い合わせ --
date_inquiry_inquiry_str=row["date_inquiry_inquiry"].isoformat(),
code_customer_inquiry=row["code_customer_inquiry"],
name_customer_inquiry=row["name_customer_inquiry"],
body_inquiry=row["body_inquiry"],
sug_num_res_route_inquiry=row["sug_num_res_route_inquiry"],
sug_name_res_route_inquiry=row["sug_name_res_route_inquiry"],
summary_inquiry=row["summary_inquiry"],
hint_res_inquiry=row["hint_res_inquiry"],
draft_auto_res_message_inquiry=row["draft_auto_res_message_inquiry"],
is_urgent_inquiry=row["is_urgent_inquiry"],
# -- 注文 --
id_order=row["id_order"],
code_order=row["code_order"],
date_order_order_str=row["date_order_order"].isoformat(),
code_customer_order=row["code_customer_order"],
name_customer_order=row["name_customer_order"],
code_member_order=row["code_member_order"],
name_member_order=row["name_member_order"],
code_item_order=row["code_item_order"],
name_item_order=row["name_item_order"],
quantity_order=row["quantity_order"],
total_price_order=row["total_price_order"],
date_delivery_yotei_order_str=row["date_delivery_yotei_order"].isoformat(),
date_delivery_order_str=row["date_delivery_order"].isoformat() if row["date_delivery_order"] else None,
is_delivered_order=row["is_delivered_order"],
date_shipped_order_str=row["date_shipped_order"].isoformat() if row["date_shipped_order"] else None,
is_shipped_order=row["is_shipped_order"],
code_item_category_order=row["code_item_category_order"],
name_item_category_order=row["name_item_category_order"],
is_delivered_on_time_order=row["is_delivered_on_time_order"]
)
for row in rows
]
return InquiryResponseSearchResult(
count=len(inquiry_response_list),
inquiry_responses=inquiry_response_list
)
except Exception as e:
# raise することで FastMCP が isError=True のエラー応答を自動生成する
# （MCPプロトコル標準の方式）
logger.error(f"get_inquiry_responses エラー: {e}")
raise
# ----------------------------------------------------------------
# update_inquiry ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class UpdateInquiryResult(BaseModel):
"""
update_inquiry ツールの戻り値。
更新が正常に完了した場合に返る。
DBエラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
success: bool = Field(description="問い合わせデータの更新が成功したらtrue。")
code_inquiry: str = Field(description="更新した問い合わせデータの問い合わせ番号。")
# ----------------------------------------------------------------
# update_inquiry ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def update_inquiry(
code_inquiry: Annotated[str, Field(
description="更新対象の問い合わせ番号。完全一致。"
)],
is_begun: Annotated[bool, Field(
description="この問い合わせは対応着手済みか。着手済みにするならtrue、未着手にするならfalse。"
)],
is_urgent: Annotated[bool, Field(
description="対応の緊急性の有無。有りにするならtrue、無しにするならfalse。"
)],
summary_inquiry: Annotated[str, Field(
description="問い合わせ本文の要約。問い合わせ者の訴えの内容を、情報を欠落させることなく、文体を簡潔に要約する。"
)],
sug_num_res_route: Annotated[int, Field(
description="対応ルートのAI提案。メール自動送信回答を提案するなら1、人間が対応することを提案するなら2。"
)],
hint_res: Annotated[Optional[str], Field(
description="人間が問い合わせに対応する場合の対応のヒント。即ち、問い合わせ対応ルートとして人間による対応がAIにより提案された場合、その対応のヒントになるもの。",
default=None
)],
draft_auto_res_message: Annotated[Optional[str], Field(
description="メール自動送信回答での回答の本文案。即ち、問い合わせ対応ルートとしてメール自動送信回答がAIにより提案された場合、その本文案。",
default=None
)]
) -> UpdateInquiryResult:
"""
健康食品の販売システムにおいて、単一の問い合わせデータを更新する。
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
# カーソルはこのtool呼び出しのスコープ内で作成・クローズする
# autocommit=True のため、明示的な conn.commit() は不要
conn = get_db_connection()
cur = conn.cursor()
try:
# DBのFunction（returns void）を呼び出す
cur.execute(
"SELECT public.update_inquiry(%s, %s, %s, %s, %s, %s, %s)",
(
code_inquiry,
is_begun,
is_urgent,
summary_inquiry,
sug_num_res_route,
hint_res,
draft_auto_res_message
)
)
finally:
cur.close()
return UpdateInquiryResult(
success=True,
code_inquiry=code_inquiry
)
except Exception as e:
# raise することで FastMCP が isError=True のエラー応答を自動生成する
# （MCPプロトコル標準の方式）
logger.error(f"update_inquiry エラー: {e}")
raise
if __name__ == "__main__":
mcp.run(
transport="streamable-http",
host="192.168.x.x", # このMCPサーバーを実行するマシンのプライベートIPアドレス
port=8002
)
MCPサーバー「sales_stat」のところでも書きましたが、
FastMCPは、「@tool」が付加された関数をMCPツールと判断し、その中身を見て、MCPサーバーの全ツールのSchemaを自動生成します。
inputSchemaだけでなく、実のあるoutputSchemaを生成させるために、各MCPツールの戻り値をBaseModelの派生クラスで詳細に定義しました。
MCPサーバー「inquiry-urg_ntf」
本稿の「MCPサーバー3」です。
緊急性の有る問い合わせ全件を、社内に通知します。
FastMCPを使用しています。
クリックして表示
| 項目 | 内容 |
|---|---|
| サーバー名 | inquiry-urg_ntf |
| ファイル名 | mcp_server_inquiry_urg_ntf.py |
| Transport | Streamable HTTP |
| 認証 | 静的Bearer Token |
MCPサーバー「sales_stat」と同じで、認証に静的Bearer Tokenを用いています。
トークンが平文でHTTPリクエストに含まれるため、本来はHTTPSで通信すべきですが、閉じられた宅内LANで使用しているだけなので、そこまではしませんでした。
ツール
ツール内部で、適したDBのFunctionを使用しています。
| 名称 | 機能 |
|---|---|
| notify_unnotified_urgent_inquiries | 緊急性の有る問い合わせデータのうち、社内未通知のもの全件を社内に通知し、社内通知済とする。 対象となった問い合わせデータの問い合わせ番号一覧を返す。 |
コード
MCPサーバー「mcp_server_inquiry_urg_ntf.py」
from fastmcp import FastMCP
import os
from typing import List
from pydantic import BaseModel, Field
import logging
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import AsyncIterator
from datetime import date
import psycopg2
import psycopg2.extras
import requests
from pathlib import Path
from dotenv import load_dotenv
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier
load_dotenv(Path(__file__).resolve().parent / ".env")
# ロギング設定（出力フォーマット追加）
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [INQUIRY-URG-NTF-SERVER] %(message)s"
)
logger = logging.getLogger("fast-mcp-server-inquiry-urg-ntf")
# DB接続をグローバル変数として保持
# ※カーソルはtool呼び出しごとに新規作成する
db_connection = None
@dataclass
class AppContext:
server_name: str = "MCP Server for notifying employees of urgent inquiries"
start_time: str = ""
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
global db_connection
try:
yield AppContext()
finally:
# 終了時：クリーンアップ
if db_connection is not None:
db_connection.close()
db_connection = None
# verifierを作り、FastMCPに渡す
_bearer_token = os.environ.get("MCP_INQUIRY_URG_NTF_BEARER_TOKEN")
if not _bearer_token:
raise ValueError("環境変数 'MCP_INQUIRY_URG_NTF_BEARER_TOKEN' が設定されていません。")
verifier = StaticTokenVerifier(
tokens={
_bearer_token: {
"client_id": "inquiry-urg-ntf-client",
"scopes": ["data:access"]
}
},
required_scopes=["data:access"]
)
mcp = FastMCP("notifying_employees_of_urgent_inquiries_server",
instructions="健康食品販売会社への問い合わせのうち、緊急対応を要する問い合わせを社内に通知する。",
lifespan=app_lifespan,
auth=verifier)
def get_db_connection():
"""
DB接続を返す。接続が無い or 切断済みの場合のみ新規作成。
カーソルは呼び出し元で都度作成すること。
"""
global db_connection
if db_connection is None or db_connection.closed:
try:
db_dsn = os.environ.get('DBDSN')
if not db_dsn:
raise ValueError("環境変数 'DBDSN' が設定されていません。")
# DSN (Data Source Name) を使用して接続
db_connection = psycopg2.connect(dsn=db_dsn, sslmode='require')
# autocommit モードを設定
db_connection.autocommit = True
except Exception as e:
raise
return db_connection
# ----------------------------------------------------------------
# notify_unnotified_urgent_inquiries ツールの戻り値 BaseModel 定義
# ----------------------------------------------------------------
class NotifyUrgentInquiriesResult(BaseModel):
"""
notify_unnotified_urgent_inquiries ツールの戻り値。
code_inquiries=[] の場合は、対象となる未通知の緊急問い合わせが0件であったことを示す（正常応答）。
DBエラー・Slack送信エラー等の例外発生時は、isError=Trueのエラー応答が返る。
"""
code_inquiries: List[str] = Field(
description="社内に通知し、通知済み状態に更新した問い合わせの問い合わせ番号一覧。0件の場合は空リスト。"
)
# ----------------------------------------------------------------
# notify_unnotified_urgent_inquiries ツール内部で使用するレコード型
# （DBから取得した、未通知の緊急問い合わせ1件分。ツールの戻り値ではない）
# ----------------------------------------------------------------
class _UnnotifiedUrgentInquiryRecord(BaseModel):
id: int
code_inquiry: str
date_inquiry: date
code_customer: str
name_customer: str
code_order: str
code_member: str
name_member: str
summary_inquiry: str
# ----------------------------------------------------------------
# notify_unnotified_urgent_inquiries ツールの内部処理（①②③）
# ----------------------------------------------------------------
def _get_unnotified_urgent_inquiries() -> List[_UnnotifiedUrgentInquiryRecord]:
"""
①緊急性が有るがまだ社内未通知の問い合わせを全件取得する。
"""
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
cur.execute("SELECT * FROM public.get_unnotified_urgent_inquiries()")
rows = cur.fetchall()
finally:
cur.close()
return [
_UnnotifiedUrgentInquiryRecord(
id=row["id"],
code_inquiry=row["code_inquiry"],
date_inquiry=row["date_inquiry"],
code_customer=row["code_customer"],
name_customer=row["name_customer"],
code_order=row["code_order"],
code_member=row["code_member"],
name_member=row["name_member"],
summary_inquiry=row["summary_inquiry"]
)
for row in rows
]
def _build_slack_message(records: List[_UnnotifiedUrgentInquiryRecord]) -> str:
"""
②未通知の緊急問い合わせ一覧を、Slack通知用のプレーンテキストに整形する。
問い合わせ間は空白行1行で区切る。
"""
blocks = []
for r in records:
block = (
"-----\n"
f"【問い合わせ番号】{r.code_inquiry}\n"
f"【問い合わせ日】{r.date_inquiry.isoformat()}\n"
f"【顧客】{r.code_customer}：{r.name_customer}\n"
f"【注文番号】{r.code_order}\n"
f"【注文を受けた社員】{r.code_member}：{r.name_member}\n"
f"【問い合わせ要約】{r.summary_inquiry}\n"
"-----"
)
blocks.append(block)
return "\n\n".join(blocks)
def _send_slack_notification(records: List[_UnnotifiedUrgentInquiryRecord]) -> None:
"""
②整形済みの未通知緊急問い合わせ一覧を、Slack（Incoming Webhook）に一度に送信する。
送信先チャンネルはWebhook URLに紐づく固定チャンネル。
"""
webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
if not webhook_url:
raise ValueError("環境変数 'SLACK_WEBHOOK_URL' が設定されていません。")
message_text = _build_slack_message(records)
response = requests.post(
webhook_url,
json={"text": message_text},
timeout=10
)
if response.status_code != 200:
raise RuntimeError(
f"Slack通知の送信に失敗（status_code={response.status_code}, body={response.text}）"
)
def _mark_unnotified_urgent_inquiries_as_notified() -> List[str]:
"""
③②で社内通知した問い合わせ（緊急性が有り社内未通知状態のままとなっている問い合わせ）を、社内通知済み状態に更新する。
更新した全問い合わせの問い合わせ番号を返す。
"""
conn = get_db_connection()
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
try:
cur.execute("SELECT * FROM public.mark_unnotified_urgent_inquiries_as_notified()")
rows = cur.fetchall()
finally:
cur.close()
return [row["code_inquiries"] for row in rows]
# ----------------------------------------------------------------
# notify_unnotified_urgent_inquiries ツール本体
# ----------------------------------------------------------------
@mcp.tool()
async def notify_unnotified_urgent_inquiries() -> NotifyUrgentInquiriesResult:
"""
緊急性が有る問い合わせのうち、まだ社内通知していないものを、社内に通知する。
そして、当該の問い合わせを社内通知済み状態に更新する。
"""
# ①緊急性が有るがまだ社内未通知の問い合わせを全件取得
try:
records = _get_unnotified_urgent_inquiries()
except Exception as e:
logger.error(f"notify_unnotified_urgent_inquiries ①未通知緊急問い合わせの取得でエラー: {e}")
raise RuntimeError(f"notify_unnotified_urgent_inquiries ①未通知緊急問い合わせの取得でエラー: {e}") from e
# 対象が0件の場合はSlack送信を行わず、即座に空リストを返す
if not records:
return NotifyUrgentInquiriesResult(code_inquiries=[])
# ②通知用に整形して、一度にSlack送信
try:
_send_slack_notification(records)
except Exception as e:
logger.error(f"notify_unnotified_urgent_inquiries ②Slack通知の送信でエラー: {e}")
raise RuntimeError(f"notify_unnotified_urgent_inquiries ②Slack通知の送信でエラー: {e}") from e
# ③緊急性が有るがまだ社内未通知の問い合わせを、社内通知済み状態に更新
try:
code_inquiries = _mark_unnotified_urgent_inquiries_as_notified()
except Exception as e:
logger.error(f"notify_unnotified_urgent_inquiries ③通知済み状態への更新でエラー: {e}")
raise RuntimeError(f"notify_unnotified_urgent_inquiries ③通知済み状態への更新でエラー: {e}") from e
return NotifyUrgentInquiriesResult(code_inquiries=code_inquiries)
if __name__ == "__main__":
mcp.run(
transport="streamable-http",
host="192.168.x.x", # このMCPサーバーを実行するマシンのプライベートIPアドレス
port=8003
)
MCPサーバー「sales_stat」のところでも書きましたが、
FastMCPは、「@tool」が付加された関数をMCPツールと判断し、その中身を見て、MCPサーバーの全ツールのSchemaを自動生成します。
inputSchemaだけでなく、実のあるoutputSchemaを生成させるために、MCPツールの戻り値をBaseModelの派生クラスで詳細に定義しました。