「Azure OpenAI でFunction Callingを試してみた」の続編として、今回は MCP (Model Context Protocol) と LangGraph を組み合わせ、 AWS DynamoDB と Azure AI SearchをまたいでFAQを検索・回答する“実用系ミニエージェント”を作ってみました。
内容は、
①MCPサーバで社内のデータソースを“標準化ツール”として公開。
②LangGraphのReActエージェントから、そのツールを呼び出して AWS DynamoDB と Azure AI Search と ローカルを横断検索。
③結果は 最も関連する1件の本文を取得して 1〜2文で要約＋出典 を返す。
というシンプルなものになります。
アーキテクチャ概要
以下は「MCP × LangGraph FAQエージェント」の全体像です。
MCPサーバが AWS / Azure / local を横断検索し、優先度（AWS > azure > local） に従って本文を取得します。
LangGraph ReActエージェント … 思考 & ツール実行の制御（ReAct）
MCPサーバ … データソースの違いを吸収し、統一インターフェース（search_faq / get_faq_by_id）を提供
AWS DynamoDB / Azure AI Search / Local ストア … FAQデータの保管先
フォールバック順 … aws > azure > local
local フォールバックの役割
開発/検証の容易さ：クラウド接続がないローカル環境でも FAQ を再現できる
冗長構成の最後の砦：AWS・Azureの両方で未ヒット/障害時に最後の候補を返せる
コードの説明（クライアント側：langchain_client.py）
1. MCPサーバとのやり取り部分
今回の仕組みでは、FAQデータ（AWS DynamoDB と Azure AI Search と ローカル に格納）にアクセスする処理を MCPサーバ 側に集約しています。
langchain_client.py 側では、そのMCPサーバを「外部ツール」として呼び出すだけの構成です。
# MCPサーバの起動パラメータ
server_params = StdioServerParameters(
command=sys.executable,
args=[str(ROOT / "mcp_server.py")],
cwd=str(ROOT),
env={**os.environ, "PYTHONUNBUFFERED": "1"},
)
ここでは、同じディレクトリにある mcp_server.py を Python で起動するように指定しています。
実際の呼び出しは mcp_invoke() 関数で行われ、MCPのツール呼び出し結果を「素のPython型」に変換して扱えるようにしています。
async def mcp_invoke(tool_name: str, arguments: Dict[str, Any]) -> Any:
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
res = await session.call_tool(tool_name, arguments=arguments)
return mcp_result_to_py(res)
2. LangGraphから利用可能なツール定義
LangGraph の ReAct エージェントが直接呼び出せるように、MCPツールを同期関数として公開しています。
今回はsearch_faq と get_faq_by_id の2つを定義しました。
@tool
def search_faq(query: str, top_k: int = 3) -> str:
"""FAQ検索を行い、上位候補をJSON文字列で返す。"""
async def _run():
res = await mcp_invoke("search_faq", {"query": query, "top_k": int(top_k)})
return json.dumps(res, ensure_ascii=False, indent=2)
return anyio.run(_run)
@tool
def get_faq_by_id(id: str) -> str:
"""FAQのIDから本文を取得してJSON文字列で返す。引数: id(str)"""
async def _run():
res = await mcp_invoke("get_faq_by_id", {"id": id})
return json.dumps(res, ensure_ascii=False, indent=2)
return anyio.run(_run)
search_faq はFAQを検索し、候補（id・本文・provider情報など）をJSONで返します。
get_faq_by_id は候補の id を指定して、本文を取り出します。
このように、MCPサーバで「データソースの違いを吸収」し、LangGraph 側では単純なツール呼び出しにしています。
3. LangGraph: Prebuilt ReAct Agent
LangGraphには「ReActエージェント」が用意されており、思考とツール実行を交互に繰り返す流れを自動で処理してくれます。
今回は、そのプリセットを利用しています。
graph = create_react_agent(
model=llm,
tools=[search_faq, get_faq_by_id],
)
ここで渡している llm は Azure OpenAI の GPTモデルです。
ReActエージェントは次の手順でFAQ回答を導きます。
-
search_faq を呼んで候補リストを得る
-
最も関連するものの id を使って get_faq_by_id を呼ぶ
-
本文を 1〜2文で要約し、最後に出典を付ける
4. 実行エントリポイント
最後に、実際の動作例として「パスワードを忘れた場合の手順は？」という質問を投げています。
if __name__ == "__main__":
print("=== Demo: FAQ 質問（LangGraph ReAct） ===")
question = "パスワードを忘れた場合の手順は？"
prompt = f"""次の質問に答えてください: 「{question}」
手順:
1) search_faq(query=質問, top_k=5) を必ず1回呼ぶ。
2) 候補一覧を確認し、内容が重複/競合する場合は provider が 'ddb' のものを優先する。
3) 最も関連する候補の id を使い、まず get_faq_by_id(id=..., provider="ddb") を呼ぶ。
もし本文が取得できなかった場合に限り、get_faq_by_id(id=..., provider="azure") を呼ぶ。
4) 回答を日本語で1〜2文にまとめ、最後に 出典(source, provider, id) を1行添える。
ツールを使わずに推測で答えないこと。"""
result = graph.invoke({"messages": [("user", prompt)]})
この部分で、LangGraphが自動的に「検索 → 本文取得 → 要約」の流れを実行してくれます。
5.実行結果
実際にスクリプトを実行すると、以下のような結果が得られました。
PS C:\Users\admin\rag-mcp-demo> python langchain_client.py
=== Demo: FAQ 質問（LangGraph ReAct） ===
--- Agent Answer ---
パスワードを忘れた場合は、ログイン画面の「パスワードをお忘れですか」からメールまたはSMSでリセットが可能です。
出典: kb_auth.md, provider: ddb, id: FAQ-003
ここでは、AWS DynamoDB側に登録されていたFAQデータが検索され、具体的な手順が1文で要約され、さらに出典情報（ファイル名・provider・ID） が添えられて返ってきていることが分かります。
このように、MCP経由で複数クラウドにまたがるFAQ検索を行い、LangGraph ReActエージェントが最終的な回答をまとめてくれています。
6.処理フロー図
今回のエージェントが FAQ に回答するまでの流れ図にすると、以下のようになります。
コードの説明（MCPサーバ側）
続いて mcp_server.py の仕組みです。
1. 環境変数と接続設定
まずは .env から接続情報を読み込みます。
Azure / AWS / local の 3種類を準備し、クラウド未接続でもローカルファイルで代替できるようにしています。
AZURE_ENDPOINT = os.getenv("AZURE_SEARCH_ENDPOINT")
AZURE_KEY = os.getenv("AZURE_SEARCH_API_KEY")
AZURE_INDEX = os.getenv("AZURE_SEARCH_INDEX")
AWS_REGION = os.getenv("AWS_REGION") or "ap-northeast-1"
DDB_TABLE = os.getenv("DDB_TABLE_NAME") or "faq_table"
LOCAL_CANDIDATES = [
os.path.join(os.path.dirname(__file__), "local_kb.json"),
]
*DynamoDB / Azure Search が使えない環境でも JSON ファイルから FAQ を検索できる
*実運用とローカル開発の両方をカバーする柔軟な構成
2.ローカルフォールバック
クラウドに接続できない場合でも、ローカルからFAQを取得できるようにしています。
def _search_local(query: str, top_k: int = 3) -> List[Dict[str, Any]]:
qn = _norm(query)
hits = []
for d in _iter_local_docs():
q = d.get("question", d.get("title", "")) or ""
a = d.get("answer", d.get("content", "")) or ""
tn = _norm(q + "\n" + a)
if qn in tn or _norm(q) == qn:
hits.append({
"id": d.get("id") or d.get("doc_id") or "UNKNOWN",
"question": q,
"snippet": a[:160],
"score": 1.0,
"source": d.get("source", "local"),
"provider": "local",
})
return hits[:top_k]
*質問テキストと回答本文を正規化して単純一致/部分一致検索
*snippet（回答の冒頭160文字）を返すので概要確認が可能
3. AWS DynamoDB / Azure Search
AWSとAzureの検索ロジックはそれぞれ独立して実装されています。
結果が取れなかった場合には、自動でローカル検索にフォールバックします。
def _search_ddb(query: str, top_k: int = 3) -> List[Dict[str, Any]]:
ddb = boto3.resource("dynamodb", region_name=AWS_REGION)
table = ddb.Table(DDB_TABLE)
resp = table.scan(ProjectionExpression="#id,#q,#a,#s",
ExpressionAttributeNames={"#id":"id","#q":"question","#a":"answer","#s":"source"},
ConsistentRead=True)
...
# 完全一致なら score=5, 部分一致なら score=2 を付与
def _search_azure(query: str, top_k: int = 3) -> List[Dict[str, Any]]:
client = SearchClient(
endpoint=AZURE_ENDPOINT,
index_name=AZURE_INDEX,
credential=AzureKeyCredential(AZURE_KEY),
)
results = client.search(search_text=query, top=top_k)
return [{
"id": doc.get("id") or "UNKNOWN",
"question": doc.get("question") or "",
"snippet": (doc.get("answer") or "")[:160],
"score": float(doc.get("@search.score", 0.0)),
"source": doc.get("source", "azure_search"),
"provider": "azure",
} for doc in results]
*DynamoDBはシンプルに scan で全件を対象にし、完全一致/部分一致でスコアを手動付与
*Azureはスコアリング済みの結果をそのまま利用
4. 検索結果の統合
AWS DynamoDBとAzure AI Searchの両方を検索し、重複を排除して上位N件を返します。
ヒットがゼロの場合はローカルを参照します。
def _search_all(query: str, top_k: int = 3) -> List[Dict[str, Any]]:
a = _search_azure(query, top_k * 2)
d = _search_ddb(query, top_k * 2)
pool = a + d
if not pool:
pool = _search_local(query, top_k)
pool.sort(key=lambda x: x.get("score", 0.0), reverse=True)
uniq, seen = [], set()
for r in pool:
key = (r.get("provider"), r.get("id"))
if key in seen: continue
uniq.append(r); seen.add(key)
if len(uniq) >= top_k: break
return uniq
*各プロバイダの結果を一度に集約
*provider × id をキーにして重複除去
*top_k まで絞り込んで返す
5. MCPツールの公開
最後に、MCP経由でエージェントから呼べるようにツールを公開しています。
@mcp.tool()
def search_faq(query: str, top_k: int = 3) -> List[Dict[str, Any]]:
"""FAQを検索（DynamoDB + Azureを横断）。"""
return _search_all(query, top_k)
@mcp.tool()
def get_faq_by_id(id: str, provider: str = "") -> Dict[str, Any]:
"""
IDでFAQ本文を取得。provider が 'ddb'/'azure'/'local'。
未指定のときは DDB→Azure→local の順でフォールバック。
"""
...
*search_faq : FAQ検索（id・question・snippet・provider を返す）
*get_faq_by_id : ID指定で本文取得（フォールバック順序付き）
このように mcp_server.py では、
・クラウドごとの差異を吸収
・フォールバックを含めた堅牢な検索処理
・LangGraphからは単純なツール呼び出しで利用可能
という設計になっています。
クライアント側からは「search_faq」「get_faq_by_id」を呼ぶだけで、裏側で DynamoDB / Azure / local のいずれかから最適なデータが返ってくるようになっています。
まとめ
今回の実装では FAQ検索に限定したシンプルなエージェントを構築し、MCP と LangGraph を組み合わせて 複数クラウド（AWS / Azure）＋ローカルを横断的に扱う基礎を確認しました。
しかし、実際の業務でエージェントが扱うデータソースやタスクが増えてくると、
「どの情報を」「どの順番で」「どの粒度で」LLM に渡すか が、結果の精度や信頼性を大きく左右します。
こうした工夫の積み重ねこそが コンテキストエンジニアリング であり、次のステップではこの観点を踏まえた応用にも取り組んでいきたいと思います。