はじめに
自然言語で何かを動かすのは、私にとって、遊びの一つです。
自然言語でDBを操作する、というのもその遊びの中の一つです。
MCP＆Function calling（GPT）を使った、自然言語でDBを読み書きする「AIコンシェルジュ」機能を作って、遊んでみました。
この画面は、よくあるチャットウィンドウ（WEB画面の右に出てくるやつ）を模したものです。
今のところ、実際にはこの画面は単独で存在します。”親”の画面はありません。
将来的には、この画面そのものを販売アプリにくっつけて、本当の「DBを読み書きをしてくれるチャットウィンドウ」にしてもいいかな、と思っています。
所詮は「遊び」なので、実装も何もかも、ただやりたいようにやり、そして書きたいように書いています。
※MCP（Model Context Protocol）については、公式ドキュメントを参照してください。
仕組みと実装
仕組み
MCPクライアントとMCPサーバーを双方、MCPのPython公式SDKを使用して、全部自前で実装しました。FastMCPは使用してません。
MCPクライアントの内部のLLMはgpt-4oです。MCPではお決まりのClaudeではありません。
MCPサーバー
MCPクライアントからの要求に従い、DBの読み書きを行います。
前提
DBヘアクセスする頻度が高くない、という想定です。
頻繁にアクセスする場合は、違った実装もあると思います。
（本稿の実装では、MCPクライアントからMCPサーバーを都度明示的に起動し、都度DB接続しているが、そうではなく、はじめからMCPサーバーを起動しておき、DB接続を保持しておく、など。そんなことできるのか、よく知りませんが。）
アクセスするDB
ある架空の農園の業務用DBです。
顧客マスタのテーブルと、販売データのテーブルがあります。
これらのテーブルを、以下のFunction（ストアドプロシージャ）を使って操作します。
「get_customers」：顧客データの収集
「get_sales」：販売データの収集
「add_sale」：販売データの登録
ツール
MCPサーバーが備えるツールは以下の3つです。
| ツール名 | 機能概要 |
|---|---|
| get_customers | 顧客データの収集 |
| get_sales | 販売データの収集 |
| add_sale | 販売データの登録 |
実装コード
import os
import json
import logging
import asyncio
from datetime import datetime
from typing import Dict, List, Optional, Any, Union
from mcp.server import Server
from mcp.server.lowlevel.server import NotificationOptions
import mcp.types as types
from pydantic import AnyUrl
from mcp.types import Resource
from dotenv import load_dotenv
import psycopg2
import psycopg2.extras
# ロギング設定（出力フォーマット追加）
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s [MCP-SERVER-SALES] %(message)s"
)
logger = logging.getLogger("mcp-server-sales")
# 環境変数の読み込み
load_dotenv()
# サーバーの初期化
server = Server("mcp-sales-server")
# リソース定義 - URIによって識別されるリソース
resources = {}
# DB接続とカーソルをグローバル変数として保持
db_connection = None
db_cursor = None
def get_db_connection():
global db_connection, db_cursor
'''有効なDB接続が無い場合のみ、新たにDB接続し、カーソルを作成'''
if db_connection is None or db_connection.closed:
（省略）
return db_connection, db_cursor
async def get_customers(code_customer: Optional[str] = None, name_str_customer: Optional[str] = None) -> Dict[str, Any]:
'''顧客データをDBから収集し、返す'''
try:
# 有効なDB接続がもし無い場合に限り、新規作成
conn, cur = get_db_connection()
# DB内Function「get_customers」を使用
cur.execute(
"SELECT * FROM get_customers(%s, %s)",
(code_customer, name_str_customer)
)
# 結果を取得
customers = cur.fetchall()
# 戻り値の辞書を生成
return {
"status": "success",
"data": {
"customers": customers,
"count": len(customers)
}
}
except Exception as e:
# エラーハンドリング
logger.error(f"顧客データ取得エラー: {e}")
return {
"status": "error",
"data": {
"error": str(e)
}
}
async def get_sales(code_customer: str, code_sales: Optional[str] = None, date_sales_begin_str: Optional[str] = None, date_sales_end_str: Optional[str] = None) -> Dict[str, Any]:
'''販売データをDBから収集し、返す'''
try:
# 有効なDB接続がもし無い場合に限り、新規作成
conn, cur = get_db_connection()
# 販売日の期間検索をする場合
# 日付文字列が指定されている場合はISO 8601フォーマットに変換
if date_sales_begin_str:
try:
# 日付の解析とISO 8601形式への変換
date_obj = datetime.strptime(date_sales_begin_str, '%Y-%m-%d')
# ISO 8601形式（YYYY-MM-DD）に変換
date_sales_begin_str = date_obj.strftime('%Y-%m-%d')
except ValueError:
logger.error(f"販売日期間の開始日を日付型に変換できませんでした: {date_sales_begin_str}")
raise ValueError(f"販売日期間の開始日の形式が不正です。: {date_sales_begin_str}")
if date_sales_end_str:
try:
# 日付の解析とISO 8601形式への変換
date_obj = datetime.strptime(date_sales_end_str, '%Y-%m-%d')
# ISO 8601形式（YYYY-MM-DD）に変換
date_sales_end_str = date_obj.strftime('%Y-%m-%d')
except ValueError:
logger.error(f"販売日期間の終了日を日付型に変換できませんでした: {date_sales_end_str}")
raise ValueError(f"販売日期間の終了日の形式が不正です。: {date_sales_end_str}")
# DB内Function「get_sales」を使用
cur.execute(
"SELECT * FROM get_sales(%s, %s, %s, %s)",
(code_customer, code_sales, date_sales_begin_str, date_sales_end_str)
)
# 結果を取得
sales_data = cur.fetchall()
# 結果が空でないことを確認
if not sales_data:
return {
"status": "error",
"data": {
"error": "販売データ取得に失敗しました"
}
}
# 最初のレコードのstatus_codeを確認
status_code = sales_data[0]['status_code']
status_message = sales_data[0]['status_message']
match status_code:
case 404:
# 顧客が存在しない場合（status_code = 404）
return {
"status": "error",
"data": {
"error": status_message
}
}
case 204:
# 検索条件に合うデータがない場合（status_code = 204）
return {
"status": "success",
"data": {
"sales": sales_data,
"count": 0
}
}
case _:
# その他、検索条件に合うデータがある場合（status_code = 200）
return {
"status": "success",
"data": {
"sales": sales_data,
"count": len(sales_data)
}
}
except Exception as e:
# エラーハンドリング
logger.error(f"販売データ取得エラー: {e}")
return {
"status": "error",
"data": {
"error": str(e)
}
}
async def add_sale(code_customer: str, products: str, price: int, date_sales_str: Optional[str] = None) -> Dict[str, Any]:
'''販売データをDBに登録する'''
try:
# 有効なDB接続がもし無い場合に限り、新規作成
conn, cur = get_db_connection()
# 日付文字列の処理
if date_sales_str:
try:
# 日付の解析とISO 8601形式への変換
date_obj = datetime.strptime(date_sales_str, '%Y-%m-%d')
# ISO 8601形式（YYYY-MM-DD）に変換
date_sales_str = date_obj.strftime('%Y-%m-%d')
except ValueError:
logger.error(f"販売日を日付型に変換できませんでした: {date_sales_str}")
raise ValueError(f"販売日の形式が不正です。: {date_sales_str}")
else:
# 指定がない場合は現在の日付をISO 8601形式で設定
date_sales_str = datetime.now().strftime('%Y-%m-%d')
# DB内Function「add_sale」を使用
cur.execute(
"SELECT * FROM add_sale(%s, %s, %s, %s);",
(code_customer, products, price, date_sales_str)
)
# 結果を取得（1レコードのみ）
result = cur.fetchone()
# status_code に応じて戻り値を作成
if result['status_code'] == 404:
# 顧客が存在しない場合はエラーとして扱う
return {
"status": "error",
"data": {
"error": result['status_message']
}
}
else:
# 正常終了の場合（status_codeが201）
return {
"status": "success",
"data": {
"sales_registered": result
}
}
except Exception as e:
# エラーハンドリング
logger.error(f"販売データ登録エラー: {e}")
return {
"status": "error",
"data": {
"error": str(e)
}
}
# 全ツールの仕様のリスト
@server.list_tools()
async def list_tools() -> list[types.Tool]:
"""
利用可能なツールの仕様のリストを返す
AIエージェントでは、LLMがその仕様のリストを使用する
ユーザーの依頼を実現するために、LLMは、どのツールが適切か判断し、そのツールの引数を生成する
"""
return [
types.Tool(
name="get_customers",
description="販売アプリにおいて、顧客マスタの検索を行い、顧客データを返す。（プロンプト例：顧客コード1234の名前と住所が知りたい。）（プロンプト例：佐藤さんというお客様の顧客データが見たい。）",
inputSchema={
"type": "object",
"properties": {
"code_customer": {
"type": "string",
"description": "顧客マスタの検索条件の1つ。顧客コード。完全一致。オプション。"
},
"name_str_customer": {
"type": "string",
"description": "顧客マスタの検索条件の1つ。顧客名の一部または全部。部分一致。オプション。"
}
}
}
),
types.Tool(
name="get_sales",
description="販売アプリにおいて、販売データの検索を行い、検索結果を返す。（プロンプト例：顧客コード1234のお客様の販売実績を知りたい）（プロンプト例：顧客コード1234の先月の販売合計額は？）",
inputSchema={
"type": "object",
"properties": {
"code_customer": {
"type": "string",
"description": "販売データの必須検索条件の1つ。検索したい販売データの販売相手である顧客のコードで絞りたい場合に使用。"
},
"code_sales": {
"type": "string",
"description": "販売データの任意検索条件の1つ。検索したい販売データの販売番号で指定して検索したい場合に使用。"
},
"date_sales_begin_str": {
"type": "string",
"description": "販売データの任意検索条件の1つ。検索したい販売データの販売日を期間で指定して検索したい場合に使用。その検索期間の開始年月日。YYYY-MM-DDの形式。"
},
"date_sales_end_str": {
"type": "string",
"description": "販売データの任意検索条件の1つ。検索したい販売データの販売日を期間で指定して検索したい場合に使用。その検索期間の終了年月日。YYYY-MM-DDの形式。"
}
},
"required": ["code_customer"]
}
),
types.Tool(
name="add_sale",
description="販売アプリにおいて、販売データの登録を行い、登録された販売データを返す。（プロンプト例：販売登録。キュウリ15本とナス8個を2000円で顧客コード1234のお客様に売った。）（プロンプト例：販売登録。昨日、キュウリ15本とナス8個を2000円で売った。顧客コードは1234。）",
inputSchema={
"type": "object",
"properties": {
"code_customer": {
"type": "string",
"description": "販売相手の顧客のコード。必須。"
},
"products": {
"type": "string",
"description": "販売した物。必須。（例：キュウリ15本とナス8個）"
},
"price": {
"type": "integer",
"description": "販売価格。通貨を除いた整数部分。必須。（例：2000）"
},
"date_sales_str": {
"type": "string",
"description": "販売した年月日。YYYY-MM-DDの形式。オプション。"
}
},
"required": ["code_customer", "products", "price"]
}
)
]
# ツール呼び出し
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> list[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
"""
指定されたツールを引数を使って実行する
LLMが選択したツールと生成した引数に基づいて処理を行う
"""
result = None
if name == "get_customers":
# MCPクライアントからの要求は、顧客データの収集
code_customer = arguments.get("code_customer", None)
name_str_customer = arguments.get("name_str_customer", None)
result = await get_customers(code_customer, name_str_customer)
elif name == "get_sales":
# MCPクライアントからの要求は、販売データの収集
code_customer = arguments.get("code_customer")
code_sales = arguments.get("code_sales", None)
date_sales_begin_str = arguments.get("date_sales_begin_str", None)
date_sales_end_str = arguments.get("date_sales_end_str", None)
result = await get_sales(code_customer, code_sales, date_sales_begin_str, date_sales_end_str)
elif name == "add_sale":
# MCPクライアントからの要求は、販売データの登録
code_customer = arguments.get("code_customer")
products = arguments.get("products")
price = arguments.get("price")
date_sales_str = arguments.get("date_sales_str", None)
result = await add_sale(code_customer, products, price, date_sales_str)
else:
result = {"status": "error", "data": {"error": "不明なツール"}}
# 結果を返す
return [types.TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]
# 各リソースの付属情報のリスト
@server.list_resources()
async def list_resources_info() -> list[Resource]:
（省略）
# リソースを返す
@server.read_resource()
async def read_resource(resource_uri: AnyUrl) -> str:
（省略）
# メイン関数 - MCPサーバーが独立プロセスとしてスクリプト実行された時の事実上のエントリーポイント
async def main():
"""
MCPサーバーを実行するメイン関数
独立したプロセスとしてスクリプト実行された際のエントリーポイント
"""
# MCPサーバー実行
from mcp.server.stdio import stdio_server
try:
# MCPクライアントと標準入出力を介したやり取りができる状態にする
async with stdio_server() as (read_stream, write_stream):
from mcp.server.models import InitializationOptions
# MCPサーバーを実際に動かして、標準入出力を介した待ち受け状態を作る
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="mcp-sales-server",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={}
)
)
)
except Exception as e:
logger.error(f"MCPサーバー（販売サーバー）の起動中にエラーが発生しました: {e}")
raise
# MCPサーバーが独立プロセスとしてスクリプト実行された時のエントリーポイント
if __name__ == "__main__":
asyncio.run(main())
MCPクライアント
UI「AIコンシェルジュ」画面から、ユーザーの自然言語の要求を受け取ります。
そのユーザーの自然言語での要求をLLM（GPT）に伝え、Function callingを行わせます。
LLM（GPT）が選択したツールと生成した引数をMCPサーバーに渡し、MCPサーバーにその実行を要求します。
その実行結果をLLM（GPT）に渡して最終応答を生成させ、UI「AIコンシェルジュ」画面に渡します。
実装コード
import asyncio
import json
import os
import logging
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
from openai import AsyncOpenAI
from dotenv import load_dotenv
from datetime import datetime
# 環境変数の読み込み
load_dotenv()
env_vars = os.environ.copy()
# ロギング設定
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s [MCP-CLIENT-SALES] %(message)s"
)
logger = logging.getLogger("mcp-client-sales")
# MCPサーバーの設定（グローバル変数として定義）
SERVER_COMMAND = os.getenv("PYTHON_PATH", "python") # デフォルトは "python"
SERVER_SCRIPT = "./server_sales/mcp_server_sales.py"
SERVER_PARAMS = StdioServerParameters(
command=SERVER_COMMAND,
args=[SERVER_SCRIPT],
env=env_vars
)
LLM_MODEL_NAME ="gpt-4o"
# LLMに投入するシステムメッセージ
LLM_SYSTEM_MESSAGE = (
"あなたは、ある販売アプリケーションのAIエージェントです。ユーザーの依頼に応じて、顧客マスタの顧客データを収集したり、販売登録したり、販売データの収集をしたりします。この販売アプリケーションには関係ないプロンプトにはツールを適用しないで、無視してください。"
)
# システムメッセージに、週の定義と、現在の年月日と曜日の情報を付加。
# これをしないと、ユーザーの自然言語での要求にほぼ必ず含まれる、現時点からの相対表現（週も含む）に対応できない。
# 例）「顧客コードxxxxのお客様の先月の販売データ全件」「顧客コードxxxxのお客様に、先週水曜日、～を1000円で販売した。」
today = datetime.now()
iso_date = today.strftime("%Y-%m-%d")
weekday_ja = ["月曜日", "火曜日", "水曜日", "木曜日", "金曜日", "土曜日", "日曜日"]
weekday_str = weekday_ja[today.weekday()]
LLM_SYSTEM_MESSAGE += f"\n週は、月曜日にはじまり、日曜日に終わります。\n現在の日付は {iso_date}（{weekday_str}）です。"
# グローバル変数で状態を保持（各関数は自己完結するが、状態は共有）
available_tools = []
available_tools_gpt = [] # OpenAI形式でのツール一覧
conversation_history = []
is_initialized = False
def initialize_client():
"""
MCPクライアントを初期化し、ツール一覧を取得する
"""
global available_tools, available_tools_gpt, is_initialized
if is_initialized:
logger.info("既に初期化済みのため、再初期化をスキップします")
return True
async def async_init():
global available_tools, available_tools_gpt
# MCPサーバーとMCPで通信
# MCPサーバーのためのプロセスを開始、MCPサーバーをスクリプト実行、MCPサーバーとの標準入出力通信ルートを構築
async with stdio_client(SERVER_PARAMS) as (read, write):
# MCPサーバーとのMCP通信セッションを構築
async with ClientSession(read, write) as session:
# MCPサーバーとの初期化ハンドシェイク
await session.initialize()
# MCPサーバーからツール一覧取得
list_tool_result = await session.list_tools()
# 戻り値の型は、mcp.types.ListToolsResult
available_tools = list_tool_result.tools
# MCPサーバーから取得したMCP形式のツール一覧をOpenAI形式に変換
available_tools_gpt = [
{
"type": "function",
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
for tool in available_tools
]
# async with ClientSessionを抜け、MCPサーバーとのMCP通信セッションが自動的に遮断される
# async with stdio_clientを抜け、MCPサーバーとの標準入出力通信ルートの遮断とサーバープロセス終了が自動で行われる
return True
try:
result = False
result = asyncio.run(async_init())
is_initialized = result
return result
except Exception as e:
logger.error(f"MCPクライアント（販売アプリ）初期化処理中にエラー: {type(e).__name__}: {e}")
return False
def process_user_prompt(user_input: str) -> str:
"""
ユーザーのプロンプトを処理し、GPTにツールを選択させ、
必要なツール呼び出し・最終応答までを行う。
"""
if not is_initialized:
logger.info("MCPクライアント（販売アプリ）が未初期化のため、初期化を実行")
initialize_client()
conversation_history.append({"role": "user", "content": user_input})
async def async_process():
try:
# clientインスタンスを生成
client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
# GPTが、ユーザー依頼に対応するのに適切なツールをツール一覧から選択、引数も生成
response = await client.responses.create(
model=LLM_MODEL_NAME,
tools=available_tools_gpt,
input=[{"role": "system", "content": LLM_SYSTEM_MESSAGE}] + conversation_history
)
# response.outputはlist
# GPTからの返答responseは、「自然言語でのメッセージ」or「選択したツール」のどちらか
response_out_type = response.output[0].type
# response_out_typeが"message"なら「自然言語でのメッセージ」 → ツールは選択せず
# response_out_typeが"function_call"なら「選択したツール」 → 以下の、MCPサーバーにツールを実行させる処理に入る
# MCPサーバーとMCPで通信
# MCPサーバーのためのプロセスを開始、MCPサーバーをスクリプト実行、MCPサーバーとの標準入出力通信ルートを構築）
async with stdio_client(SERVER_PARAMS) as (read, write):
# MCPサーバーとのMCP通信セッションを構築
async with ClientSession(read, write) as session:
# MCPサーバーとの初期化ハンドシェイク
await session.initialize()
while response_out_type == "function_call":
# GPTが何らかのツールを選択
tool_calls = response.output
# GPTが選択したツールを順次実行
for call in tool_calls:
if call.type != "function_call":
continue
# 会話履歴にツール選択結果を追加
# callの型は、openai.types.responses.response_function_tool_call.ResponseFunctionToolCall
conversation_history.append(call)
# GPTが選択したツールの名前を取得
tool_name = call.name
# GPTが生成したツールの引数を取得
arguments = json.loads(call.arguments)
# GPTより採番されたcall_idを取得
call_id = call.call_id
# MCPサーバーにツールを実行させる
result_content = await session.call_tool(tool_name, arguments)
# MCPサーバーでのツール実行結果のjson文字列（欲しい答えはこれ）を戻り値result_contentから抽出
# 戻り値result_contentはmcp.types.CallToolResult型
# result_content.contentはlistで、MCPサーバーのcall_toolのシグネチャの戻り値の型list[Union[mcp.types.TextContent, mcp.types.ImageContent, mcp.types.EmbeddedResource]]そのもの
# MCPサーバー「mcp_server_sales」のcall_toolの実行結果はmcp.types.TextContentオブジェクトが1個のみのlistであることがわかっているので、以下のような簡素な抽出法でよい。
result_text = result_content.content[0].text
# JSONパースできる場合、status確認（ツール実行時にエラーが起こったかどうか）
# しかし、is_errorフラグそのものには使いみちは無いので、やらなくてもいいかも。
# ツール実行時のエラーには、指定された顧客コードが存在しない、というのがあり、必ずしもこのユーザー依頼対応の順次処理を止めるべきものとは限らない。
try:
result_json = json.loads(result_text)
is_error = (result_json.get("status") == "error")
e_msg_tool = result_json.get("data").get("error") if is_error else ""
except Exception:
logger.warning("結果文字列をJSONに変換できず")
# ツール実行結果文字列のJSON変換に失敗しているので、ツール実行時にエラー発生とみなす
is_error = True
e_msg_tool = "ツール実行時に不明なエラー"
if is_error:
# ツール実行時にエラーが起こっても処理は止めない。
# ツール実行のこのターン（forループの）でエラーが起こっても、次のターンに影響を及ぼさない。
logger.warning(f"ツール実行時のエラー:{e_msg_tool}")
# 会話履歴にツール実行結果を追加
conversation_history.append({
"type": "function_call_output",
"call_id": call_id,
"output": result_text
})
# for ループを抜けた。GPTにより選択されていた全ツールの実行を完了した。
# 最終応答の生成 or 次のFunction callのループ
# ＜最終応答の生成＞
# ユーザーの自然言語での要求が、実行すべき全ツールを確定させるものの場合
# このresponseのcreateでは、全ツールの実行が完了したので、最終応答の生成となる。
# response.output[0].typeは"message"になる。
# ＜次のFunction callのループ＞
# ユーザーの自然言語での要求が、実行すべきツールを全確定させない（あるツールの実行結果で、次のツールが決まる）ものの場合
# もし、そのツールの実行結果が、さらなるツールの実行を要求する時、
# このresponseのcreateでは、最終応答の生成ではなく、次のFunction callのループとなる。
# response.output[0].typeは"function_call"になる。
response = await client.responses.create(
model=LLM_MODEL_NAME,
tools=available_tools_gpt,
input=[{"role": "system", "content": LLM_SYSTEM_MESSAGE}] + conversation_history
)
response_out_type = response.output[0].type
# response_out_typeが"message"なら、responseには最終応答が入っている。whileを抜ける。
# response_out_typeが"function_call"なら、responseには次に実行すべきツールが入っている。whileループをもう一回行う。
# while response_out_type == "function_call": ループを抜けた
# async with ClientSessionを抜け、MCPサーバーとのMCP通信セッションが自動的に遮断される
# async with stdio_clientを抜け、MCPサーバーとの標準入出力通信ルートの遮断とサーバープロセス終了が自動で行われる
#最終応答の抽出
final_text = response.output_text
# 会話履歴に最終応答を追加
conversation_history.append({"role": "assistant", "content": final_text})
return final_text
except Exception as e:
logger.error(f"プロンプト処理中にエラー発生: {type(e).__name__}: {e}")
return f"[MCP] エラーが発生しました: {str(e)}"
try:
return asyncio.run(async_process())
except Exception as e:
logger.error(f"asyncio実行エラー: {type(e).__name__}: {e}")
return f"[MCP] 実行エラー: {str(e)}"
GPTにシステムメッセージで現在年月日と週の定義を与える
LLMは、「今がいつか」を知りません。
考えてみれば当たり前で、そんなの学習データには無いからです。
アプリケーションとしての「ChatGPT」は今がいつかを知っていますが、それはアプリケーション「ChatGPT」の話であって、API経由で扱うLLM「GPT」の話ではないです。
一方で、ユーザーの自然言語での要求には、ほぼ必ず、現時点からの相対表現が含まれます。
週と曜日を使用した相対表現も当然あり得ます。
例）「顧客コードxxxxのお客様の先月の販売データ全件を見せて。」
例）「販売登録して。顧客コードxxxxのお客様に、先週水曜日、～を1000円で販売した。」
具体的な年月日が使われる方が、むしろ少ないはずです。
このような相対表現をGPTが解釈して、ツールの年月日の引数を生成する場合、
・今は何年何月何日の何曜日か
・「週」は何曜日から何曜日までか
という情報がGPT自身に必要となります。
ですので、それらをシステムメッセージに含めて、GPTに提供します。
# LLMに投入するシステムメッセージ
LLM_SYSTEM_MESSAGE = (
"あなたは、ある販売アプリケーションのAIエージェントです。ユーザーの依頼に応じて、顧客マスタの顧客データを収集したり、販売登録したり、販売データの収集をしたりします。この販売アプリケーションには関係ないプロンプトにはツールを適用しないで、無視してください。"
)
# システムメッセージに、週の定義と、現在の年月日と曜日の情報を付加。
# これをしないと、ユーザーの自然言語での要求にほぼ必ず含まれる、現時点からの相対表現（週も含む）に対応できない。
# 例）「顧客コードxxxxのお客様の先月の販売データ全件」「顧客コードxxxxのお客様に、先週水曜日、～を1000円で販売した。」
today = datetime.now()
iso_date = today.strftime("%Y-%m-%d")
weekday_ja = ["月曜日", "火曜日", "水曜日", "木曜日", "金曜日", "土曜日", "日曜日"]
weekday_str = weekday_ja[today.weekday()]
LLM_SYSTEM_MESSAGE += f"\n週は、月曜日にはじまり、日曜日に終わります。\n現在の日付は {iso_date}（{weekday_str}）です。"
LLMにClaudeは使わない（私は）
1つのプロンプト中に複数の要求がある（＝複数回のツールの実行を必要とする）場合、LLMにClaudeを使っていると、以下のようにフラストレーションが溜まるやり取りを強制されます（2025年3月時点）。
例）プロンプト「顧客コード0001のお客様の電話番号と、今までの販売件数を教えて。」
この後どうなるか、ClaudeとGPTとで比較してみます。
ただし、Claudeについては今までの経験からの推測です。きっとこうなります。
| やること（上から順に） | Claudeの場合(今までの経験からの推測) | GPTの場合 |
|---|---|---|
| ①LLMによるツール選択 | 顧客データを収集する「get_customers」 を選択。 |
顧客データを収集する「get_customers」 販売データを収集する「get_sales」 を選択。 |
| ②MCPサーバーのツール実行 | 「get_customers」を実行して、結果をMCPクライアントに返す。 | 「get_customers」を実行して、結果をMCPクライアントに返す。 「get_sales」を実行して、結果をMCPクライアントに返す。 |
| ③LLMからの応答 | 「顧客コード0001の佐藤太郎様の電話番号はXXXXです。」 | 「顧客コード0001の佐藤太郎様の電話番号はXXXXです。」 「顧客コード0001の佐藤太郎様の販売件数は……です。」 という最終応答で終わる。 |
| ④LLMによるツール選択 | 販売データを収集する「get_sales」 を選択。 |
（既に終了。） |
| ⑤LLMからの意向不明の確認 | 「次に、顧客コード0001の販売データを取得します。」 と、ユーザーにわざわざGOサインのプロンプト投入を求める。 |
（既に終了。） |
| ⑥ユーザーからの怒りのGOサインのプロンプト投入 |
「やれって言ってるんだから、いちいち聞き返して流れを止めずにさっさとやれよ！」 と、ユーザーはやむをえずGOサインのプロンプトを投入する必要あり。 |
（既に終了。） |
| ⑦MCPサーバーのツール実行 | 「get_sales」を実行して、結果をMCPクライアントに返す。 | （既に終了。） |
| ⑧LLMからの応答 | 「顧客コード0001の佐藤太郎様の販売件数は……です。」 という最終応答で終わる。 |
（既に終了。） |
Claudeがなぜこのような挙動をするのか、についてなのですが、どうもそういうものらしいです。
私はERPの上流工程を10年以上やってましたが、ユーザビリティの観点では、こんな仕様はあり得ないです。
Anthropicも、それなりの理由があって、敢えてこうしているのだとは思いますけどね。
ツール一覧をOpenAI形式へ変換
MCPサーバーから取得したツール一覧はMCP形式です（汎用性を考えて、MCPサーバー側でそうしました）。
このままGPTに突っ込んでもエラーを吐くだけなので、OpenAI形式に変換する必要があります。
# MCPサーバーから取得したMCP形式のツール一覧をOpenAI形式に変換
available_tools_gpt = [
{
"type": "function",
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
for tool in available_tools
]
MCPとFunction callingの合わせ技 詳説
ユーザーの自然言語プロンプト投入から、ユーザーへの最終応答返却までは、以下の流れになります。
主要部分は、OpenAIの「Function calling」です。
LLMの判断材料になるツール一覧の取得と、実際のツールの実行、の2ヶ所で、MCPサーバーが絡みます。
１．事前にMCPサーバーからツール一覧を取得し、保持
MCPクライアント初期化処理のinitialize_client関数において、MCPサーバーからツール一覧を取得し、OpenAI形式に変換します。
～～～～～～
２．ユーザーがUIにプロンプトを投入し、MCPクライアントへとそのプロンプトが渡される
UIからMCPクライアントのprocess_user_prompt関数に、ユーザーのプロンプトが送られます。
プロンプトは、「販売登録して。顧客コードは1234，〇年×月△日に、キュウリ5本を720円で売った。そして今月のこのお客様の販売実績はどうなった？」とします。
３．LLMに、ユーザーのプロンプトを実現するツールの選択と引数の生成をさせる
ユーザーのプロンプトを会話履歴に追加します。
そして、保持してあるOpenAI形式のツール一覧とその会話履歴をGPTに投入し、以下の2つをさせます。
・ツールの選択
・選択したツールの引数の生成
GPTは、「ユーザーのプロンプト」と「ツール一覧」を照合し、ツールの選択と引数の生成をします。
ここでは、GPTは、以下のようにツールを選択し、それらツールの引数を生成したとします。
| GPTが選択したツール | GPTが生成した引数 |
|---|---|
| add_sale | 顧客コード＝1234，販売日=〇年×月△日、販売物＝キュウリ5本、販売価格＝720 |
| get_sales | 顧客コード＝1234，販売期間=〇年×月 |
ユーザーのプロンプトは、
「販売登録して。顧客コードは1234，〇年×月△日に、キュウリ5本を720円で売った。そして今月のこのお客様の販売実績はどうなった？」
で、1つのプロンプトの中に2つの要求が含まれています。
なので、GPTは、ツールを2回分選択したわけです。
この３を、OpenAI用語では「Function calling」、Anthropic用語では「Tool use」と言います。どちらも同じ意味です。
LLMが、プロンプトとともに提供された関数（Function）情報一覧から、適切な関数を選択し、その関数の引数も生成することを言います。
４．MCPサーバーに、ツールを順次実行させる
以下のサイクルを、選択されたツール全てについて、繰り返します。
選択されたツールとその引数を会話履歴に追加
↓
選択されたツールを、引数を添えて、MCPサーバーに実行させる
↓
そのツールの実行結果を会話履歴に追加
今の例では、以下のようになります。
ツール「add_sale」が選択され、
引数は「顧客コード＝1234，販売日=〇年×月△日、販売物＝キュウリ5本、販売価格＝720」
であることを会話履歴に追加
↓
MCPサーバーに、ツール「add_sale」を
引数（顧客コード＝1234，販売日=〇年×月△日、販売物＝キュウリ5本、販売価格＝720）を添えて実行させる
↓
ツール「add_sale」の実行結果（成功＆新規登録された販売データ）
を会話履歴に追加
↓ 次のツールのサイクル
ツール「get_sales」が選択され、引数は「顧客コード＝1234，販売期間=〇年×月」
であることを会話履歴に追加
↓
MCPサーバーに、ツール「get_sales」を引数（顧客コード＝1234，販売期間=〇年×月）を添えて実行させる
↓
ツール「get_sales」の実行結果（顧客コード＝1234で販売期間=〇年×月の販売データ全件）
を会話履歴に追加
５．LLMに、ユーザーへの最終応答を生成させる
GPTに、会話履歴をもとに、自然言語での最終応答を生成させます。
この時点での会話履歴は以下の通りです（上に行くほど新しい）。
| 履歴の順番 （大きいほど 新しい） |
履歴の中身 |
|---|---|
| 5 | ツール「get_sales」の実行結果。 顧客コード＝1234で販売期間=〇年×月の販売データ全件。 |
| 4 | ツール「get_sales」が選択された。 引数は「顧客コード＝1234，販売期間=〇年×月」。 |
| 3 | ツール「add_sale」の実行結果。 成功＆新規登録された販売データ。 |
| 2 | ツール「add_sale」が選択された。 引数は「顧客コード＝1234，販売日=〇年×月△日、販売物＝キュウリ5本、販売価格＝720」。 |
| 1 | ユーザーのプロンプト。 「販売登録して。顧客コードは1234，〇年×月△日に、キュウリ5本を720円で売った。そして今月のこのお客様の販売実績はどうなった？」 |
GPTは、これまでの会話履歴を踏まえて、最終応答を生成します。
ここでは、以下の最終応答を生成したとします。
「顧客コード：1234，販売日：〇年×月△日、商品：キュウリ5本、価格：720円、販売番号：xxxxで、販売登録しました。」
「顧客コード：1234、販売日：〇年×月 の販売実績は・・・・・です。」
このGPTの最終応答を会話履歴に追加し、やり取りを完結させます。
結果、この時点での会話履歴は以下の通りとなります（上に行くほど新しい）。
| 履歴の順番 （大きいほど 新しい） |
履歴の中身 |
|---|---|
| 6 | GPTの最終応答。 「顧客コード：1234，販売日：〇年×月△日、商品：キュウリ5本、価格：720円、販売番号：xxxxで、販売登録しました。」 「顧客コード：1234、販売日：〇年×月 の販売実績は・・・・・です。」 |
| 5 | ツール「get_sales」の実行結果。 顧客コード＝1234で販売期間=〇年×月の販売データ全件。 |
| 4 | ツール「get_sales」が選択された。 引数は「顧客コード＝1234，販売期間=〇年×月」。 |
| 3 | ツール「add_sale」の実行結果。 成功＆新規登録された販売データ。 |
| 2 | ツール「add_sale」が選択された。 引数は「顧客コード＝1234，販売日=〇年×月△日、販売物＝キュウリ5本、販売価格＝720」。 |
| 1 | ユーザーのプロンプト。 「販売登録して。顧客コードは1234，〇年×月△日に、キュウリ5本を720円で売った。そして今月のこのお客様の販売実績はどうなった？」 |
この会話履歴の状態で、次の新たなプロンプトに対処していきます。
６．UIに最終応答を返す
MCPクライアントからUIに最終応答を返します。
MCPクライアントのprocess_user_prompt関数の戻り値が最終応答です。
遊んでみる
能書きはもうたくさんなので、早速遊んでみます。
「あなた」（ユーザー）と「AIコンシェルジュ」とのやり取りは、画面キャプチャを貼るのではなく、文字列で書き起こします。
あなた:
佐藤さんの顧客コードを教えて。
AIコンシェルジュ:
佐藤さんという名前の顧客は以下の通りです：
1. 佐藤太郎
- 顧客コード: 0001
- 住所: 東京都新宿区西新宿3丁目ヒカリビル12階
- 電話: 03-1234-5678
2. 佐藤次郎
- 顧客コード: 0009
- 住所: 京都府京都市下京区烏丸通河原町スモータル館302
- 電話: 075-666-7777
どちらの佐藤さんが該当しますか？
「佐藤」という顧客は複数いるので、AIコンシェルジュは、ちゃんと全員の情報を返してきました。
あなた:
顧客コード0009と0010の、先週の販売データを見せて。
AIコンシェルジュ:
顧客コード0009の先週の販売データは以下の通りです：
- 日付: 2025年05月10日
- 顧客: 佐藤次郎
- 商品: キュウリ15本
- 価格: 1050円
顧客コード0010の先週の販売データはありませんでした。
2人の顧客の販売データの収集なので、内部的には、販売データ収集のツール選択が2回分、それに対するツール実行も同じく2回行なわれた、となります。
ログを確認しましたが、「先週」という時間軸上の相対表現にGPTはちゃんと対応し（事前対処が必要）、販売期間検索の引数2025/5/5～2025/5/11を正しく生成していました。
顧客コード0010については、この週の販売実績が無いので、「先週の販売データはありませんでした。」と回答されています。
あなた:
顧客コード0002の先月と今月の2ヶ月間の販売金額合計は？
AIコンシェルジュ:
顧客コード0002のお客様、John Smithさんの先月と今月の2ヶ月間の販売合計金額は以下の通りです：
- 合計金額：2,180円
### 販売内訳
1. 2025年04月19日: トマト2個とかぼちゃ1個 - 340円
2. 2025年04月22日: トマト8個 - 640円
3. 2025年05月02日: かぼちゃ2個 - 240円
4. 2025年05月07日: トマト12個 - 960円
ログを確認しましたが、「先月と今月の2ヶ月間」という時間軸上の相対表現にGPTはちゃんと対応し（事前対処が必要）、販売期間検索の引数2025/4/1～2025/5/31を正しく生成していました。
見やすいように、そこそこの粒度で「販売内訳」を列挙してくれたのはスグレモノ。ただしいつもこう気が利くわけではありません。
「合計金額：2,180円」は、ツールの実行結果（販売データレコード単体が複数のリストになっている）を受け取ったGPTが、最終応答生成時にその場で計算したものです。
Function callingでは、Function（ここではツールの事）の実行結果の解釈と加工を、実行結果を受け取ったLLMにどこまでやらせるか、常に気にする必要があります。
このケースのような単純な集計程度ならLLMにやらせてもいいですが、複雑な計算ならFunction（ツール）にやらせる必要があり、その方法を別途考えなければいけません。
しかし何より重要なのは、顧客コード0002のJohn Smithさんは、トマトとかぼちゃが好きだった、ということです。それがわかったのは、AIコンシェルジュの、気の利いた「販売内訳」列挙のおかげです。
良い仕事するね、AIコンシェルジュ（中身はgpt-4o）。
あなた:
顧客コード0002のお客様の先月の販売実績が欲しい。それから、山田様の顧客コードが知りたい。
AIコンシェルジュ:
### 顧客コード0002の先月の販売実績
- **2025年04月19日**
- 販売番号: S20250419112931
- 商品: トマト2個とかぼちゃ1個
- 価格: 340円
- **2025年04月22日**
- 販売番号: S20250422101931
- 商品: トマト8個
- 価格: 640円
### 山田様の顧客コード
- 顧客コード: 0003
- 名前: 山田花子
- 住所: 北海道札幌市中央区大通西4丁目ファンタジアA101
- 電話番号: 011-987-6543
販売データ収集と顧客データ収集という、全く異なる2つの要求が1個のプロンプト中でなされています。異なる2つのツール実行が必要で、それらを1プロンプトの最終応答までに一括して行っています。
GPTだからできるのであって、Claudeではきっとこうはなりません。
GPTは、「先月」という相対表現にも正しく対応しています（事前対処が必要）。
あなた:
販売登録して。顧客コード0016のお客様に、先週火曜日、かぼちゃ4個を600円で売った。
AIコンシェルジュ:
販売登録が完了しました。内容は以下の通りです：
- 顧客: 松本理恵さん (顧客コード: 0016)
- 日付: 2025年05月06日 (先週火曜日)
- 商品: かぼちゃ4個
- 価格: 600円
ついにDB書き込みである「販売登録」です。
GPTは、「先週火曜日」という相対表現にも正しく対応し、「2025年05月06日」という販売年月日の引数を正しく生成しています（事前対処が必要）。
あなた:
顧客コード0014の今月の販売件数がまだ0件の場合、急遽、販売登録を行う。ナス20個を1800円で販売。
AIコンシェルジュ:
顧客コード0014の今月の販売件数が0件だったため、新たに販売を登録しました。
斉藤明様にナス20個を1800円で販売しました。販売登録は成功しました。
あなた:
顧客コード0014の今月の販売データを見せて。
AIコンシェルジュ:
今月の顧客コード0014（斉藤明様）の販売データは以下の通りです：
- **販売日**: 2025年05月15日
- **販売番号**: S20250515132801
- **商品**: ナス20個
- **価格**: 1800円
以上の1件が登録されています。
「顧客コード0014の今月の販売件数がまだ0件の場合、急遽、販売登録を行う。ナス20個を1800円で販売。」
これは架空取引ですね。別にいいんです、この農園自体が架空だから。
これは今までのとは少し毛色が違うもので、あるツールの実行結果の条件分岐により、次の実行ツールが決まる、というものです。
顧客コード0014の今月の販売件数を探るために、GPTは販売データ収集ツール「get_sales」を選択し、MCPサーバーがそれを実行。
↓
販売データ収集ツール「get_sales」の実行結果は「0件」だった。
↓
GPTは、その実行結果「0件」を見て初めて、販売登録ツール「add_sale」を次の実行ツールとして選択し、MCPサーバーがそれを実行。
という流れになります。
一応、確認のため、
「顧客コード0014の今月の販売データを見せて。」
と、追加要求をしてみました。
その結果、ちゃんと販売登録されていることがわかりました。販売番号があるので間違いないです。
GPTは、「今月」という相対表現にも正しく対応しています（事前対処が必要）。
「遊び」を超えて ～ 基幹業務システムへの適用を考える
ここまでは、私の勝手な「遊び」の話でした。
が、私はERPの上流工程を10年以上やっていたので、遊びながらもどうしても
「この「AIコンシェルジュ」のようなものを基幹業務システムに増設するにはどうしたらいいか？」
「そもそもどういうユースケースがあるのか？やって意味あるのか？」
などと考えてしまっていました。
そこで、ここでは、
「基幹業務システムにAIコンシェルジュのようなものを増設する」
を、雑感レベルで考えたいです。
- 「AIコンシェルジュ」ではなく、実態に即した「チャットボット」という、人口に膾炙した言葉を使います。
- 皆さんが普段お勤めの組織で使用している基幹業務システムにくっつく「チャットボット」を想定します。
-
Function callingを発動してDBを操作するユースケースに限定します。
チャットボットでよく聞く要望「社内文書などのドメイン知識の検索と解釈」はここでは扱わないことにします。
以降は相当、端折ります。
「遊び」という本稿の趣旨とは異なるもので、長大にしたくないからです。
チャットボットでのデータの検索・閲覧・集計について
これが基幹業務システム付きのチャットボットならではのユースケースであろう。
チャットボットは、大したことない情報を手早く調べるのに向いている
例えば、ある顧客企業の電話番号を入金担当の経理社員が知りたい時、いちいち基幹業務システムの顧客マスタを開けて検索して・・・なんてするより、
チャットボットで
「顧客コードxxxxの電話番号は？」
と、サッと聞いてしまった方が格段に楽となる。
上記はマスタデータだが、業務データだって、簡単に収集できるものなら別に良い。
例えば、
「商品コードxxxxの在庫は今何個？」
「案件番号xxxxの現時点での想定利益率は？」
など、有用なユースケースはいくらでも出てくるだろう。
ただし、Function callingの構造である「LLMがFunction（MCPのツールの事）を選ぶ」という観点からすると、全部に対応するのは不可能、と思う（Functionの数があまりにも多いと、LLMによる誤選択が頻発するだろう）。
重い業務データを扱う場合の、既存のシステム画面とチャットボットの「棲み分け」について
棲み分けの軸は、出力の「視覚的情報量」
重い業務データの例として、
「今年四半期の、部門ごとの損益」
を取り上げる。
「今年四半期の、部門ごとの損益の根拠資料」
なら、
「全部門の全案件の損益（売上とコストも含め）」
が出力であろう。
この視覚的情報量は膨大なもので、大きな視覚領域で一覧性を以て閲覧するものである。
既存の専用システム画面や、そこから出力されるPDFなどのレポートで見たいと思うはずである。
チャットボットは著しく不適である。
一方で、
「今年四半期の、部門ごとの損益のサマリー」
なら、
「全部門の最終損益（売上とコストも含め）」
が出力であろう。
＜イメージ＞
部門A：売上（￥1,000,000） 費用（￥800,000） 損益（￥200,000） 利益率（20%）
部門B：売上（￥2,000,000） 費用（￥1,800,000） 損益（￥200,000） 利益率（10%）
部門C：売上（￥3,000,000） 費用（￥3,400,000） 損益（－￥400,000） 利益率（－）
この程度の視覚的情報量なら、チャットボットでの有用なユースケースになるのではないか。
（ただこの場合でも、せいぜい「3」部門程度ではないか。）
プロンプト例：
「今年四半期の、XY事業部の部門ごとの最終損益が欲しい。各部門の売上、費用、損益、利益率を表示して。」
実装の注意点としては、チャットボットからcallするFunctionは、サマリーそのものを戻り値とするべき、ということか。
前述の全案件のデータそのものを戻した場合、実行結果を受け取ったLLMは、最終応答生成時にそこから集計せねばならない。LLMがそこまで正しくやってくれるか、全くもって不明である。
それから、非機能要件としてのパフォーマンスの問題がある。
ここは一概に言えないので、
扱いたいデータの収集と集計にかかる時間、
「チャットボット」というI/F上でユーザーがどれほど待てるか、
個々に検討する必要があるだろう。
チャットボットでは、対応するパラメーターを絞る
チャットボットでデータ読み込みに使用するパラメーター（検索条件や集計の軸）は、チャットボットでもそれがあった方が利便性が確実に高まるものに限定する。絞りに絞る。
自然言語のプロンプトなんて、そこに収容できるパラメーターの数はたかが知れている。
般若心経のような長大なプロンプトなんて、めんどくさがって誰も投入しない。
忘れてはいけない「権限」という視点
どの組織にも「内部統制」というものが必ず存在する（モロにそういう名前でなくても）。
既存の基幹業務システムには、「権限」という形で、それが盛り込まれているはずである。
例）
・管理部門以外の人間は、個人情報テンコ盛りの社員マスタを閲覧すらできない
・経理部門の人間のみ全員の経費精算データを見ることができるが、それ以外の人間は自分と部下の分しか見ることができない
・他事業部の案件は一切見ることができない（存在すら関知させないくらい徹底した秘匿）
など。
基幹業務システムにチャットボットをくっつける場合、同様に「権限」に従う必要がある。
例えば、以下のようなことは、基幹業務システムと同様、防がないといけない。
例）
・管理部門以外の人間が、チャットボット経由で社員マスタから他の社員の個人情報を取得する
・他人の経費精算のデータをチャットボット経由で取得する
・（前述の例）他事業部の部門の損益のサマリーをチャットボット経由で取得する
など。
対処としては、
- チャットボットで取得できる情報そのものを絞る
例えば、社員マスタの個々の社員情報取得なんてチャットボットでできなくても支障がないからチャットボットでは対応しない、など。 - チャットボットでcallされるFunction内で、callした人間にこのデータを収集する権限があるか判断
ログインユーザーをFunctionの引数にしてFunction内で権限判定し、権限が無ければエラーを返す、など。 - チャットボットでcallされるFunction内で、callした人間にこのFunctionを起動する権限があるか判断
同様に、ログインユーザーをFunctionの引数にしてFunction内で権限判定し、権限が無ければエラーを返す、など。
チャットボットでのデータ登録について
既存の専用システム画面ではなく、あのちっぽけなチャットボット内で、敢えて自然言語を使用して登録したくなるようなユースケースは、「無いと思われる」。
曲がりなりにも”業務”データなのだから項目はそれなりに多いはず。
自然言語で登録するのなら、長大な般若心経のようなプロンプトになるだろう。
多くの項目をチャットボットの対応範囲から除外すればいいのだろう（ことごとくデフォルト値にする）が、
そこまでして「チャットボット」で「自然言語」での「登録」をしたくなるデータやユースケースってありえるのか？
また、ユーザーのプロンプトの「質」とLLMの「解釈力」の双方に依存するので、誤登録のリスクがある。