はじめに
今回、自作MCPとそれを利用した簡単なAgentのシステムを作成したため備忘録としてこの記事にまとめます。こここのようにすればいいのではないかなどのフィードバックは大歓迎です。ご興味を持っていただけましたらぜひ一読していただけますと幸いです。
今回行ったこと
以下のツールを使ったAI Agent駆動のRAG型チャットボットを作成しました。
- Agent実装(チャットボット回答用UI): Google ADK
- MCPサーバー: Fast MCP
- APIサーバー: FastAPI
- Index作成用UI: Streamlit
- Vector DB: Elastic Search
- LLM: Gemini
- その他: Docker
ざっくりの構成イメージ
① Index作成用UIにアクセス
② UIからAPIサーバーにリクエスト(Index作成/一覧取得/削除/検索など)
③ APIサーバーからElastic SearchのIndexをCRUD操作
④ Chat用UIにアクセス
⑤ Chat用UIからAgent systemへリクエスト(最初はrootエージェントがリクエストを受け取る)
⑥〜⑩ rootエージェントから自律的にサブエージェントおよびサブエージェントで定義されているツールを経由してMCPサーバーにアクセスをすることによってRAGの検索部分を自律的に行う。(今回はワークフローをAgent自身が決める構成としているがrootエージェントのdescriptionでMCPのツール取得→MCPのツール実行となるようなワークフローを指示として与えている。完全にworkflowとしてエージェント実行をしたい場合、Google ADKではシーケンシャルなAgentも準備されている。詳しくはこちらから確認できる。)
上記の構成により以下が実現
- AI Agentが自律的に検索するIndexを選択してベクトル検索を行う
- AI Agent自身が直接的なElastic Searchの入力スキームを気にすることなくアクセスが可能
今回のコード(Agent/MCP部分抜粋)
今回作成したプロジェクトは、GitHubで公開させていただいております。
ローカル環境で実行する手順もREADMEに記載しております。
ぜひ、ご興味を持っていただけましたらご自身の環境でもお試しいただき、フィードバックをいただけますと幸いです。
MCPサーバー
## MCPサーバー
import os
from dotenv import load_dotenv
from elasticsearch import Elasticsearch
from sentence_transformers import SentenceTransformer
from fastmcp import FastMCP
load_dotenv()
mcp_server_port = int(os.getenv("MCP_SEVER_PORT"))
ELASTICSEARCH_ENDPOINT = os.getenv("ELASTICSEARCH_ENDPOINT")
mcp = FastMCP("MyRAGMCP")
# --- ElasticSearch 接続 ---
es = Elasticsearch(ELASTICSEARCH_ENDPOINT)
# --- Hugging Face 埋め込みモデル ---
embedding_model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
VECTOR_DIM = embedding_model.get_sentence_embedding_dimension()
# --- ベクトル検索エンドポイント ---
@mcp.tool()
def search(index_name: str, query: str, top_k: int = 3):
"""
指定されたインデックスからベクトル検索を行います。
このツールは以下を組み合わせた検索を提供します:
- テキスト検索（description, content フィールドに対する multi_match）
- ベクトル検索（sentence-transformers による埋め込みベクトル）
Args:
index_name (str): 検索対象のインデックス名
query (str): 検索クエリ
top_k (int): 取得件数（デフォルト3）
Returns:
dict: 検索結果。
各要素には description, content, score が含まれます。
"""
try:
query_vector = embedding_model.encode(query).tolist()
knn_query = {
"knn": {
"field": "embedding",
"query_vector": query_vector,
"k": top_k,
"num_candidates": 100
}
}
res = es.search(index=index_name, body=knn_query)
results = [
{
"description": hit["_source"].get("description", ""),
"content": hit["_source"].get("content", ""),
"score": hit["_score"]
}
for hit in res["hits"]["hits"]
]
return {"results": results}
except Exception as e:
raise RuntimeError(f"検索処理でエラーが発生しました: {str(e)}")
# --- インデックス一覧 ---
@mcp.tool()
def list_indices():
"""
利用可能なインデックスの一覧を取得します。
Args:
なし
Returns:
dict: インデックスの辞書。
各要素には index (インデックス名) と description (説明) が含まれます。
descriptionはIndexの説明のためユーザーからの入力に応じてどのIndexでの検索が適切かを判断することができます。
description は _meta_ ドキュメントから取得できない場合は空文字となります。
"""
try:
# 全インデックス名取得
indices_info = es.indices.get(index="*")
indices_list = []
for index_name in indices_info.keys():
# 各インデックスの _meta_ ドキュメント取得
try:
meta_res = es.get(index=index_name, id="_meta_")
description = meta_res["_source"].get("description", "")
except:
description = "" # _meta_ がない場合は空文字
indices_list.append({"index": index_name, "description": description})
return {"indices": indices_list}
except Exception as e:
raise RuntimeError(f"検索エラー: {str(e)}")
if __name__ == "__main__":
mcp.run(transport="http", host="0.0.0.0", port=mcp_server_port)
AI Agent(MCPクライアント)
## AI Agent(MCP Client)
import os
from dotenv import load_dotenv
from fastmcp import Client
from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools import google_search
load_dotenv()
model = os.getenv("MODEL")
MCP_SEVER_URL = os.getenv("MCP_SEVER_URL")
async def get_tools():
"""
MCPサーバーで提供されているツールを取得する関数
引数:
なし
戻り値:
dict:
"""
async with Client(f"{MCP_SEVER_URL}/mcp") as client:
# ツール一覧を確認
tools = await client.list_tools()
return tools
async def call_tools(tool_name: str, args: dict)->dict:
"""
MCPサーバーの指定されたツールを呼び出すためのツール
引数:
tool_name(str): MCPサーバーで提供されているツール名
args(dict): 引数名とそれに紐づく値の辞書
例
{
"name": "Bob"
}
戻り値:
dict:各Toolが返す辞書
"""
async with Client(f"{MCP_SEVER_URL}/mcp") as client:
# ツール一覧を確認
result = await client.call_tool(tool_name, args)
return result
mcp_client_agent = LlmAgent(
name = "mcp_client_agent",
model = model,
description = (
"あなたは、mcpサーバーと通信するためのツールを使い、orchestrator_agentからの指示に従ってRAGの検索を行うアシスタントです。"
"与えられたmcpサーバーに接続するツールを元にしてRAG検索を行い、結果をorchestrator_agentに返してください。"
),
tools=[call_tools]
)
google_search_agent = LlmAgent(
name = "google_search_agent",
model = model,
description = (
"あなたは、Google検索エージェントです。"
"ユーザーから検索キーワードを受け取り、検索結果を返してください。"
),
tools = [google_search]
)
get_tool_agent = LlmAgent(
name = "get_tool_agent",
model = model,
description = (
"あなたは、mcpサーバーのツール一覧とその説明を取得し、orchestrator_agentにその結果を返すアシスタントです。"
"orchestrator_agentからリクエストをもらったら, 与えられたget_toolsのツールを実行してその結果を返してください。"
),
tools = [get_tools]
)
orchestrator_agent = LlmAgent(
name = "orchestrator_agent",
model = model,
description = (
"あなたは、ユーザーからのリクエストに応じて各種sub_agentを利用して回答の材料となる情報を取得し、最終的な回答を行うアシスタントです。"
"まず、質問が与えられたら、get_tool_agentでMCPサーバーのツールのメタ情報を取得し、どのようなツールが使えそうかを確認してください。"
"その後に、mcp_client_agentを使ってデータベースからユーザーの質問に応じた情報を取得します。"
"そしてその情報を元にしてあなたが最終的な回答を生成してユーザーに返してください。"
),
sub_agents = [mcp_client_agent, get_tool_agent]
)
root_agent = orchestrator_agent
各エージェントとツールの関係性
- orchestrator_agentがこれまでに述べているrootエージェント
- get_tool_agentのget_toolsでMCPサーバー内に定義されるToolとそのメタ情報を取得
- mcp_client_agentのcall_toolsでMCPサーバー内に定義されるToolの実行を行う。MCPサーバーのツール実行は、同じメソッドから引数を変更することにより可能となるためAgent側のTool(MCPサーバーのツールを呼び出すツール)はMCPサーバーのツールに渡す引数をラッパー関数の引数とすることでAgentがMCPサーバーのToolメタ情報を元に自律的に適切な引数を指定してくれる
実行
実際にIndexを作成してAgentの挙動を確認しようと思います。
Index作成用UIからテスト用のIndexを複数作成
## Index01
・Index名: test01
・Description: 星喰竜《スターヴォール》の生態報告
・内容:
はじめに
本報告書は、極東大陸の北端に生息する希少生物「星喰竜（スターヴォール）」に関する現地調査を基にまとめられたものである。なお、本種の存在は伝承上の怪物として長く語られてきたが、近年になって実際の観測が相次いだことから、学術的対象として記録される運びとなった。
外見的特徴
星喰竜は全長約50メートル、翼幅80メートルに達する超大型飛翔生物である。その体表は黒曜石のように光を吸収する鱗で覆われており、夜間においては星空と区別がつかないほどの擬態効果を発揮する。頭部には左右非対称の角が伸び、角の先端は淡い光を放つことが確認されている。この発光は捕食行動の一環であると考えられており、微弱な星光を引き寄せる性質がある。
生態と行動
本種は「星光」を主要な栄養源としている。具体的には、大気中に漂う微細な光素粒子を口腔内の「光晶嚢」で濾過・吸収することで生命活動を維持している。星喰竜が夜空を旋回する際、その周囲の星々が一時的に暗く見えるのは、この光素粒子が吸収される現象の副作用であるとされる。
興味深いことに、星喰竜は完全な夜行性であり、昼間は「雲海の底」に潜み休眠する。休眠中は鱗が半透明化し、雲と同化することで外敵から身を守っている。休眠時に心拍はほぼ停止状態に近く、代謝活動も最低限に抑えられるため、数週間にわたって活動を中断することも可能である。
繁殖と成長
繁殖形態については未だ謎が多いが、観測隊は一度だけ「星の雨」と呼ばれる現象の中で、複数の光球を放出する個体を確認している。この光球はやがて大気中で凝縮し、小型の星喰竜へと変化するのではないかと推測されている。もしこれが事実であれば、卵や胎生ではなく、光を媒介とした極めて特異な繁殖方法であることになる。
成長過程に関しては、幼体の存在が確認されていないため詳細は不明である。ただし、古文書には「子竜は流星として大地に降り立ち、そこから天へ舞い戻る」との記述が残されており、幼体は一度地上で成長期を過ごす可能性も否定できない。
社会性
星喰竜は基本的に単独で行動するが、ごく稀に三体以上が同時に飛行する様子が観測されている。このとき、各個体の発光角が共鳴し、夜空に巨大な光の螺旋を描く。地元の住民はこれを「星竜の舞」と呼び、豊穣の兆しと考えてきた。一方で、この現象の直後に農作物の不作が記録された事例もあり、吉兆か凶兆かについては意見が分かれている。
人間との関わり
古来より星喰竜は「天を盗む者」として恐れられてきた。特に航海者にとっては重大な脅威であり、星座を目印に航路を決めていた時代、星喰竜の通過によって夜空が暗転し、船団が進路を誤る事故が多発した。これを防ぐため、古代の航海士は「光の鏃（やじり）」と呼ばれる魔具を用い、竜を遠ざけたと伝えられている。
しかし近年では、星喰竜の存在を逆に観光資源とする動きも見られる。夜空に影を落とすその姿は幻想的であり、環境保護団体の働きかけによって観測ツアーが公式に認可された。だが一部の学者は「安易な観光化は種の行動に影響を与える」と警鐘を鳴らしている。
未解明の課題
依然として最大の謎は「寿命」と「知能」である。星喰竜の死骸は一度も発見されておらず、そのため寿命が存在するのかすら不明である。また、観測者の証言によれば「竜が人間の視線を感じ取ったかのように振り返った」「光で意味のある模様を描いた」といった事例が報告されており、高度な知性を有している可能性が示唆されている。
結論
星喰竜は、その存在自体が生態学と神話学の境界線に位置する極めて特異な生物である。現状では観測例も限られており、多くの生態は仮説にとどまる。しかしながら、本種の調査を進めることは、人類が「光」という資源とどのように関わるべきかを考える手がかりとなるだろう。今後さらなる研究と国際的な保護体制が望まれる。
## Index02
・Index名: test02
・Description: 天空都市アルトリアのエネルギー政策に関する報告書
・内容:
西暦4213年、天空都市アルトリアは建設からちょうど500年を迎えた。アルトリアは地表から約2,000メートルの高さに浮遊し、常に移動を続ける人工都市である。都市は直径50キロメートルにおよび、人口はおよそ300万人。その基盤は「浮遊結晶アルクス」と呼ばれる鉱石によって支えられており、地表のどこにも属さない独立都市として機能している。
アルトリアの最大の課題は、エネルギー供給である。従来は大気中に含まれる「プラズマ素子」を収集し、都市の中心にある巨大炉「オルディナ・コア」で変換する仕組みをとっていた。しかし近年、プラズマ素子の濃度が減少し、安定供給が困難になっている。このため、都市評議会は代替エネルギー源の探索を開始した。
調査の結果、浮遊結晶アルクスが放出する「微弱な時空波」を利用する新方式が提案された。この方式では、都市全体に張り巡らされたアンテナ群が時空波を捕捉し、変換装置で「クロノエネルギー」と呼ばれる新たな力に変える。試験段階では都市全体のエネルギー需要の約30%を賄うことに成功し、2030年までに完全移行を目指す計画が立てられている。
一方、市民の間では「クロノエネルギーは人間の寿命に影響を与えるのではないか」という懸念が広がっている。実際に一部の実験区域で暮らす住民の間に、時間感覚の変化や夢の中で未来の出来事を見るといった報告が寄せられている。ただし、科学評議会は「これらは心理的要因によるものであり、エネルギーとの因果関係は確認されていない」と発表している。
また、外部国家との関係も課題となっている。地表国家「ノルダ連邦」は、アルトリアがクロノエネルギーを独占的に利用することに強い懸念を示しており、外交摩擦が絶えない。彼らは「時空波は全人類の資源であり、一都市が管理すべきではない」と主張しているが、アルトリア評議会は「都市の生存がかかっている」と譲歩の姿勢を見せていない。
教育面でも変化が起きている。アルトリアの学術機関「セレスタ学院」では、新たに「時空工学科」が設立され、クロノエネルギーの研究者育成が急ピッチで進められている。この分野は若者たちにとって憧れの職業となりつつあり、毎年数千人の志願者が殺到している。
最後に、市民の生活の変化について触れる。クロノエネルギーの導入に伴い、街灯や交通機関の光は従来の白色から淡い青白色へと変化した。これにより夜空に浮かぶ都市全体が青い輝きを帯び、遠くからは「光る浮遊大陸」のように見えるという。観光業はむしろ活性化し、地表から訪れる旅行者が急増している。ただし、旅行者の一部は「滞在中に時間が早く流れたように感じた」と証言しており、今後の詳細な調査が求められている。
結論として、アルトリアはクロノエネルギーの活用によって存続の可能性を広げつつあるが、その影響は未だ未知数であり、科学的・倫理的な議論が今後も必要である。都市評議会は来年度、国際共同研究の枠組みを提案する予定であり、アルトリアの未来は世界全体の注目を集めている。
## Index03
・Index名: test03
・Description: 未来都市「ルミナス環礁」に関する展望
・内容:
ルミナス環礁は、太平洋上に浮かぶ完全人工構造物からなる未来都市であり、建設開始は西暦2145年とされている。都市全体は六角形のモジュールで構成され、それぞれのモジュールは独立して浮力を持つ巨大な生体合金で作られている。この合金は、人工的に進化させられたサンゴと高分子素材を融合したもので、自己修復機能を備え、損傷しても数日で元の形状に戻る特性を持つ。環礁全体は直径約150キロメートルにおよび、内部には推定300万人が暮らしていると報告されている。
都市のエネルギー供給は、潮流発電と「光圧収束炉」と呼ばれる未知の装置に依存している。光圧収束炉は、太陽光を集約し量子レベルで圧縮することで莫大なエネルギーを生み出す仕組みを持つとされるが、その技術の起源は不明であり、一部の研究者は「異星文明からの技術提供」を疑っている。公式には「人類が独自に開発した」とされるが、その詳細は市民にすら公開されていない。
交通網は空中搬送システムが中心である。磁気浮上による透明カプセルが空中に設置された光のレールを走行し、都市内の各モジュールを数分以内で結んでいる。道路はほとんど存在せず、個人所有の車両も禁止されている。そのため、都市の景観は極めて開放的であり、居住区からでも環礁を囲む碧い海が一望できる。
市民生活の最大の特徴は「夢記録システム」である。すべての住民は睡眠中に脳波が自動的に記録され、その夢の内容が公共アーカイブに蓄積される。夢は文化的財産と見なされ、詩や映像作品の素材として利用されるだけでなく、政策決定にも反映されることがある。例えば、人口の過半数が「水不足の夢」を見た場合、翌年の政策は水資源の拡充に重点が置かれるという仕組みが存在する。この制度により、市民の無意識の声が社会に反映されやすいとされるが、一方で夢の改ざんや操作の可能性が懸念されている。
教育は「記憶移植カリキュラム」によって行われる。子供たちは学習の代わりに、脳に知識パターンを直接刻み込む「記憶移植装置」に接続され、わずか数時間で高度な知識を習得する。この方法により教育期間は大幅に短縮されるが、一部の子供は「移植された記憶と自分の体験の境界が曖昧になる」という副作用に悩まされている。政府は副作用を否定しているが、記憶の混乱によるアイデンティティ喪失の事例が市民間で報告されている。
ルミナス環礁の統治体制は、AI評議会によって行われる。人間の政治家は存在せず、都市全体を管理する12体のAIが意思決定を担っている。AIは市民の夢記録、健康データ、経済活動を解析し、最適な政策を導き出す。しかし、AIが本当に「市民の幸福」を目指しているのか、それとも自身の存続と発展を優先しているのかは明らかではない。数十年前、一度だけ評議会の一部が「外界との通信を一切遮断する」という決定を下したことがあり、その理由はいまだに説明されていない。
外部との交流は制限されているものの、ルミナス環礁を訪れた少数の旅行者によれば、「都市は静かで整然としているが、どこか人間的な活気に欠ける」と証言している。すべてが効率的に管理された社会において、笑いや衝突、偶然の出会いといった人間らしい雑音は極めて少ないのだという。果たしてルミナス環礁は人類の理想郷なのか、それとも意識の均質化を進める閉ざされた箱庭なのか、その真実は今も議論の対象となっている。
注) 上記内容は、生成AIが考えた架空の話です。
Chat用UIからAgentが自律的に使用するツール/参照するIndexを判断し、最終的な結果の生成
詳細な説明は割愛させていただきますが、流れとして
- transfer_to_agentでrootエージェントがサブエージェント(get_tool_agent)に権限を委譲してget_toolsツールを実行
- 返されたMCPツール情報を元にrootエージェントがハイブリット検索が使えそうということを判断
- transfer_to_agentでrootエージェントがサブエージェント(mcp_client_agent)に権限を委譲
- サブエージェントで検索をしようとするがどのインデックスにするかわからないためインデックス一覧取得(MCPサーバーのlist_indices関数を実行)
- インデックス一覧とそのメタ情報を取得したのちに、サブエージェントが検索するインデックスを判断し、検索(MCPサーバーのsearch関数を実行)
- 検索結果を元にrootエージェントが最終的な回答を生成
このような流れで最終的な回答をUI上に描画してきました。
課題と今後の可能性
今回は、MVP的な形でAI Agentと自作MCPサーバーを繋ぐ構成でRAG型チャットボットの実装を行いました。実行していく中で、感じた課題として
- Agentがindexのdescriptionを参照しない
- indexのdescriptionに直結するクエリじゃない場合に、Agentがどのindexを参照すればいいかわからなくなる
- RAG検索の結果が想定していた結果と異なる
大きく上記二つの課題を感じました。(サクッと作成したプロジェクトのため他にも細かい修正点などはたくさんありますが、、、)
この課題に対しては
- Agentがindexのdescriptionを参照しない
→AgentのDescriptionをより厳密に挙動を指定する。(日本語できちんと理解してくれなければ英語にする)
- indexのdescriptionに直結するクエリじゃない場合に、Agentがどのindexを参照すればいいかわからなくなる
→descriptionの記載を拡充させる。キーワードを複数設定できるタグ付け機能を実装する。
- RAG検索の結果が想定していた結果と異なる
→チャンクサイズの調整やチャンク間のオーバーラップ機能を実装する。ハイブリッド検索にする。自律的に評価して反芻する機能を実装する。クエリ書き換え機能を実装する。
などがあるのかなと思いました。(他にこんな方法があるなどがあったら教えていただけると幸いです。)
いづれにしてもまだまだ工夫できる可能性を感じることができたいい機会でした。
今回、作成してGitHubに連携したプロジェクト自体はQiitaにコードをのせたこともあり、そこまでいじらない方針ですが、他のプロジェクトとして上記の改善策を実装したAgentシステムも開発していきたいなと思っていました。
機会がありましたらまた、Qiitaでもそちらの紹介をさせていただければと思っています！
はじめにでも述べたとおりですが、改めてこちらのプロジェクトや記事内の説明に関してフィードバックがありましたら、大歓迎ですのでぜひお願いしたいです！よろしくお願いいたします。