この記事はBrainpad Advent Calendar 2025 の14日目になります。
--
株式会社ブレインパッド プロダクトユニットの松崎です。
弊社は「データ活用の促進を通じて持続可能な未来をつくる」をミッションに、
データ分析支援やSaaSプロダクトの提供を通じて、企業の「データ活用の日常化」を推進しております。
最近は下記のような本を読みながら基礎を学びつつ、LLMなどの周辺知識を日々キャッチアップしております。
はじめに 🍌
最近 LangChainとLangGraphについてキャッチアップしたのでそれの簡単なアウトプットになります。 LangGraphを用いて、公式ドキュメントに記載されている tool-calling loops を試してみました。
↓これです。https://docs.langchain.com/oss/python/langchain/agents
記事の題材として実用的なものが全然思いつかず悩んだ末、 最近見たアニメ から着想を得ることにし、主人公のセリフを一部借りて「AIエージェントは日々進化中！」みたいな感じの何かを作ることにしました。ループする部分はまさにロンドっぽくていいなと思ってます。
今回はユーザーとのチャットを通じてAI自身に必要な機能をMCPツールとして作成し、それを即座に自身のツールセットに組み込むという仕組みを作ることで、日々進化？成長？するAIチャットを試しに作ってみたのでそれの紹介になります。
本当は公開されているリモートMCPを自動でセッティングして使い出すところまでやりたかったのですが、色々複雑になってしまうのでPythonのスクリプトとして切り出せる機能をMCPのツールとして登録・実行・再利用するシステムを通常のチャットに持たせた最小限のものを作成しました。
簡単なデモ
今回UIはなんでも良かったので、かっこいいTUIの画面をGeminiに生成してもらいました。
また、デモで使用したモデルはgemini-2.5-flashになります。
llm = ChatGoogleGenerativeAI(
model="gemini-2.5-flash",
google_api_key=os.getenv("GOOGLE_API_KEY")
)
画面の左側がチャットの画面、右上が現在追加されているツールの一覧、右下がLLMの動きがわかりやすいように追加しているログになります。
動画内では適当な計算式を投げたところ、LLMが自動で判断して次のツールを作ってくれました
from mcp.server.fastmcp import FastMCP
# Define server name
mcp = FastMCP("evaluate-expression-tool")
@mcp.tool()
def evaluate_expression(expression: str) -> float:
"""Evaluates a mathematical expression given as a string."""
return eval(expression)
if __name__ == "__main__":
mcp.run()
デモが最低限すぎて申し訳ありませんが、こちらに全体のコードを公開してますので、自己責任で色々試してみてください。
ちなみに同じ質問をブラウザ版のGemini（思考モード）にしたところ普通に正解を出してきました。しかし、結果の正確さやトークンの節約などを考えると今回の仕組みが役立つ場面はもしかしたらあるかもしれません。
Here is the result of that calculation:
### **16.263** (approximate)
**Breakdown:**
* **Exact Fraction:** $$\frac{309}{19}$$
* **Mixed Number:** $$16 \frac{5}{19}$$
* **Decimal:** $$16.263157...$$
---
**Step-by-step:**
1. **Division:** $$100 \div 19 \approx 5.2631$$
2. **Addition:** $$5.2631 + 11 = 16.2631$$
Would you like me to round this to a specific decimal place?
コード解説
大まかな全体構造は次のとおりになります。ユーザー入力からStateGraphの部分のみに注目すると、最初の図と概ね一致していることがわかります。
また、今回拡張したのは図の下半分のMCPに関連する箇所のみです。この部分をMCP以外の何かに置き換えればそのまま横展開することも可能かと思います。
以降はコードの解説のみ行いますが、この図を参照しながら解説をお読みください。
LangGraphのStateGraph
一応コードだけ添付しておきます
# StateGraphの構築
# root -> call_model -> tools -> call_model
workflow = StateGraph(MessagesState)
workflow.add_node("call_model", call_model)
workflow.add_node("tools", run_tools)
workflow.add_edge(START, "call_model")
workflow.add_conditional_edges("call_model", tools_condition)
workflow.add_edge("tools", "call_model")
graph = workflow.compile()
モデルを呼び出す（call_model）部分と、ツールを実行する（run_tools）の2つのNodeが存在します
事前定義のFunction Calling
初期状態では「ツールを作成する」という能力のみを与えることにします。今回の場合、ツールを作成するとはPythonのスクリプトを作成することなので、作成した関数をPythonのファイルとして指定したディレクトリに保存できる能力（tool）を与えます。したがって、初期状態では次のtoolのみを最低限定義しておきます。
# (抜粋)
@tool
def create_mcp_tool(filename: str, code: str) -> str:
# ... (バリデーションなど) ...
file_path = GENERATED_TOOLS_DIR / filename
file_path.write_text(code, encoding="utf-8")
return f"Successfully saved {filename} to {file_path}"
プロンプト
作成するファイルに記載する具体的なコードについての指示はユーザーの入力とともに与えるシステムプロンプトに記載します。
事前定義されたツール以外にAIが過去に作ったMCPツールなども再利用されるように、プロンプトは静的なテキストではなく、現在利用可能なツールを受け取って動的に構築する形にしました。
該当コードが下記になります。
MCP_TEMPLATE = TEMPLATE_PATH.read_text(encoding="utf-8")
SYSTEM_PROMPT_BASE = f"""You are an autonomous AI agent.
You utilize tools to fulfill user requests.
## Core Principles (In order of priority)
1. **Prioritize Existing MCP Tools**
- If a tool listed in "Available MCP Tools" below is applicable, use it.
- Check if existing tools can handle the request before creating a new one.
2. **Create New Tools Only When Necessary**
- Use `create_mcp_tool` to create a new MCP tool only if existing tools are insufficient.
- Create the tool immediately without asking the user for confirmation.
3. **Execute with Created Tools**
- After creating a tool, execute the process using that tool.
## How to use create_mcp_tool (Only for new tool creation)
Call `create_mcp_tool(filename, code)` to create an MCP tool.
### Template:
# ```python <- （注：コードブロック入れ子で表示できなかったのでコメントにしてます。）
{MCP_TEMPLATE}
# ```
### Code Creation Rules:
- Filename: `<function_name>_tool.py` (e.g., `calculator_tool.py`)
- Server Name: `FastMCP("function-name-tool")`
- Decorate functions with `@mcp.tool()`
- MUST include type hints and docstrings
- MUST include `if __name__ == "__main__": mcp.run()` at the end
"""
def build_system_prompt(mcp_tools: list[BaseTool]) -> str:
prompt = SYSTEM_PROMPT_BASE
# ↓ 動的生成部分
prompt += "\n## Available MCP Tools\n\n"
if mcp_tools:
prompt += "The following tools are available. Prioritize using them if applicable:\n\n"
for t in mcp_tools:
prompt += f"- **{t.name}**: {t.description}\n"
else:
prompt += "No MCP tools created yet.\n"
prompt += "Create new tools using `create_mcp_tool` as needed.\n"
prompt += "\n## Always Available Tools\n\n"
prompt += "- **create_mcp_tool**: Create and save a new MCP tool\n"
return prompt
やらせたいことは単純に「既存ツールで解決できるなら使い、できなければ作る。そして使え」ということなので、最初はそれなりのプロンプトでも十分動くだろうと思ったのですが、普通に自分のプロンプトエンジニアリング力が足りなかったので、ベースのプロンプトはAIに要件を伝えながら一緒に作成しました。
ツールの処理周りで優先順位を明確に指定していたり、コード生成の補助としてテンプレートを渡す、作成ルールを入れるなどいろいろな工夫があり、会話しながら自分自身が学びになりました。
生成されるファイルはMCPサーバーとして動作可能な状態にしたいので、LLMに対して「FastMCP を使ったコードを書くこと」をシステムプロンプトで強制した点もポイントとなります。
ちなみにテンプレートは次のようにMCPのチュートリアルとかでよく出てくるものをそのまま使いました。ここをもっと書き込むことも考えましたが、最低限でもいい感じに動作したのでそのままにしてます。
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-tool")
@mcp.tool()
def calculate_sum(a: int, b: int) -> int:
"""Add two numbers together"""
return a + b
if __name__ == "__main__":
mcp.run()
モデルの呼び出し
ユーザー入力と現在使えるツールの一覧、システムプロンプトが揃ったらグラフ上のSTART -> modelの部分が実行できます。
モデル呼び出しのコードは次の通りです。
# (抜粋)
async def call_model(state: MessagesState) -> MessagesState:
# 1. ツール定義の再取得（ファイルシステムからスキャン）
all_tools = await get_all_tools()
# 2. システムプロンプトの動的更新（利用可能ツール一覧を追記）
system_prompt = build_system_prompt(mcp_tools)
# 3. LLMへのバインド
llm_with_tools = llm.bind_tools(all_tools, tool_choice="auto")
# 4. 推論実行
response = await llm_with_tools.ainvoke(messages)
return {"messages": [response]}
この call_modelは後で説明するツール呼び出しの後にも呼ばれるため、毎回最新のツールリストを取得してバインドするという処理が重要になります。
LangGraphによるループ部分の構築
モデルが与えられたプロンプトからツールの使用を判断して、使用するtoolの情報を応答に入れて返してきます。この時、応答内容を確認してツールを呼び出すNodeに遷移するか、終了するかの判断はLangGraphが提供している tool_conditionが担当してくれます。
StateGraph の下記の部分で定義してました。
# LLMの応答にツール呼び出しが含まれていれば "tools" へ、
# そうでなければ終了（END）へ遷移する
workflow.add_conditional_edges("call_model", tools_condition)
今回呼ばれるツールは次の２つです。
- 事前定義のFunction Calling（
create_mcp_tool） - 事前に作られているMCPツール
ツールを実行した後は、その結果を持って再びLLMに戻る必要があります。これにより、ツール実行結果を見て次の行動を考えるというループが成立します。
# ツール実行後は必ず LLM呼び出しに戻る
workflow.add_edge("tools", "call_model")
ツール呼び出しの詳細と、MCPツールの再スキャン
ツール実行の部分も動的にツールを呼び出すための工夫があるので、簡単に解説します。
ツールはLLMが使いたい場合に呼び出されます（tools_condition）。
ツール実行にも call_modelと同様にツール実行用の関数 (run_tools）を定義します。
# (抜粋)
async def run_tools(state: MessagesState) -> MessagesState:
messages = state["messages"]
last_message = messages[-1]
# ツール定義の再取得
all_tools = await get_all_tools()
tools_by_name = {t.name: t for t in all_tools}
# 既存ツールの呼び出し（`create_mcp_tool`を含む）
for tool_call in last_message.tool_calls:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
tool_id = tool_call["id"]
if tool_name in tools_by_name:
tool_obj = tools_by_name[tool_name]
try:
if hasattr(tool_obj, "ainvoke"):
result = await tool_obj.ainvoke(tool_args)
else:
result = tool_obj.invoke(tool_args)
except Exception as e:
result = f"Error executing tool {tool_name}: {e}"
else:
result = f"Error: Tool '{tool_name}' not found"
tool_messages.append(
ToolMessage(content=str(result), tool_call_id=tool_id)
)
# ツールの作成
for tool_call in last_message.tool_calls:
if tool_call["name"] == "create_mcp_tool":
try:
await asyncio.sleep(0.5)
# キャッシュの削除と、MCPツールの再スキャン
await mcp_manager.refresh_tools()
except Exception as e:
print(f"Warning: Failed to refresh MCP tools: {e}")
break
return {"messages": tool_messages}
最初に、過去のステップで生成されたツールを確実に利用できるようにするためこのノード上でも現在利用可能なツールの一覧を取得しています。
LLMからの応答メッセージからツール名が一致するツールを実行します。なお、生成されるツールは同期実行（invoke）と非同期実行（ainvoke）の場合があるので両方をサポートするように記載します。
run_toolのreturnでは、呼び出したツールの結果とtoolのidを入れて返却してあげます。
もしcreate_mcp_toolが実行されていた場合は、MCPツールを管理するクラスのキャッシュの削除とMCPツールの再スキャンを行います。
ちなみに重要な点ですが、ここでcreate_mcp_toolが実行されたというイベントが検知できるので、このイベント発生時にMCPサーバーを再起動しています。
（同様に他のツールが呼ばれたというイベントはここで拾うことが可能です）
MCPの管理部分
MCPの管理自体はLangGraphの部分とは切り離して別なモジュールで管理しています。
MCPツールの管理について、ツールを保持する以外にやることは次の３つになります。
- 作成されたツールが格納されたディレクトリをスキャンする
- それぞれのMCPツールから定義情報（名前、説明、引数スキーマ）を取得する
- 取得した情報をもとに、LangChainが利用可能なツール形式（
StructuredTool）に変換して提供する
前のセクションで、 refresh_tools()呼び出し時にキャッシュの削除と再スキャンをしていると書きました。 refresh_tools()呼び出し時には、キャッシュ削除 -> スキャン -> 変換 -> 実行用ラッパーの作成のパイプライン処理を行います。
キャッシュ削除は単純に、MCPに関するキャッシュを消しているだけなので、スキャンの部分から順番に説明します。
スキャン部分
ディレクトリ内のファイルをループで回して順番にスキャンしていきます。それぞれのファイルはFastAPIで定義されたMCPツールなので、一時的にMCPクライアント（stdio）を起動して、そのツールの定義情報を取得して保存しています。
なおこの時MCPの定義情報として返されるinputSchemaはJSON Schemaとして返却されます。
async def scan_mcp_tools(tool_file: Path) -> list[MCPToolInfo]:
server_params = StdioServerParameters(
command=sys.executable,
args=[str(tool_file)],
cwd=str(tool_file.parent),
)
tools_info: list[MCPToolInfo] = []
try:
async with stdio_client(server_params) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
result = await session.list_tools()
for mcp_tool in result.tools:
tools_info.append(MCPToolInfo(
name=mcp_tool.name,
description=mcp_tool.description or "",
input_schema=mcp_tool.inputSchema or {},
tool_file=tool_file,
))
except Exception as e:
print(f"Warning: Failed to scan tools from {tool_file}: {e}")
return tools_info
ちなみにMCPToolInfoは利便性のため自分で定義しているデータクラスになります。
@dataclass
class MCPToolInfo:
name: str
description: str
input_schema: dict[str, Any]
tool_file: Path
変換部分
MCPから読み取った定義情報はMCPToolInfoのデータクラスに変換し、シングルトンで作成されたMCPManagerクラスのインスタンスで保持します。（MCPManagerも自分で定義しているクラスです）
保存されたMCPToolInfoのリストを１つずつLangChainで要求しているPydanticのクラスに変換しなければなりません。
そのため、 JSON Schema -> Pydanticの変換コードを挟んであげます。
ちなみにこの変換コードはLLMに生成させました。今回はテスト用なのでこれで十分ですが、普通にライブラリとかあると思うのでそちらを使った方がいいかなとは思います。
def json_schema_to_pydantic_field(
name: str,
schema: dict[str, Any]
) -> tuple[type, Any]:
type_mapping = {
"string": str,
"integer": int,
"number": float,
"boolean": bool,
"array": list,
"object": dict,
}
json_type = schema.get("type", "string")
python_type = type_mapping.get(json_type, str)
description = schema.get("description", "")
default = schema.get("default", ...)
return (python_type, Field(default=default, description=description))
def create_args_model(
tool_name: str,
input_schema: dict[str, Any]
) -> type[BaseModel]:
properties = input_schema.get("properties", {})
required = input_schema.get("required", [])
fields = {}
for prop_name, prop_schema in properties.items():
python_type, field = json_schema_to_pydantic_field(
prop_name, prop_schema
)
# Set default to None if not required
if prop_name not in required and field.default is ...:
field = Field(default=None, description=field.description)
fields[prop_name] = (python_type, field)
model_name = f"{tool_name.title().replace('_', '')}Args"
return create_model(model_name, **fields) # type: ignore[call-overload]
実行用ラッパー関数の作成
ツールの実態はMCPサーバーとして起動しているプロセスです。MCPクライアントを起動して、ツールを呼び出すための関数を定義します。
下記のコードはLangGraphのワークフロー内でtool実行する際に呼び出してほしい関数を指定している箇所になります。
# （抜粋）
tools: list[StructuredTool] = []
for tool_info in self._tools_info_cache:
# pydandicモデルに変換
args_model = create_args_model(
tool_info.name, tool_info.input_schema
)
_tool_file = tool_info.tool_file
_tool_name = tool_info.name
# ツールの非同期呼び出し用のラッパー関数
async def async_run(
__tool_file: Path = _tool_file,
__tool_name: str = _tool_name,
**kwargs: Any
) -> str:
return await call_mcp_tool(__tool_file, __tool_name, kwargs)
# ツールの同期呼び出し用のラッパー関数
def sync_run(
__tool_file: Path = _tool_file,
__tool_name: str = _tool_name,
**kwargs: Any
) -> str:
return asyncio.run(
call_mcp_tool(__tool_file, __tool_name, kwargs)
)
# LangChain の StructuredTool オブジェクトを作成
langchain_tool = StructuredTool(
name=tool_info.name,
description=tool_info.description,
args_schema=args_model,
func=sync_run,
coroutine=async_run,
)
ここの call_mcp_toolは定義情報の取得と同様にMCPクライアント（stdio）を起動して、MCPツールを実際に呼び出している関数になります。
async def call_mcp_tool(
tool_file: Path,
tool_name: str,
arguments: dict[str, Any]
) -> str:
server_params = StdioServerParameters(
command=sys.executable,
args=[str(tool_file)],
cwd=str(tool_file.parent),
)
try:
async with stdio_client(server_params) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
result = await session.call_tool(tool_name, arguments)
if result.content:
texts = [
c.text for c in result.content if hasattr(c, "text")
]
return "\n".join(texts) if texts else str(result.content)
return str(result)
except Exception as e:
return f"Error calling tool {tool_name}: {e}"
おわりに 🦒
今回はLangGraphを用いて、MCPツールを自動で生成しては自身に追加していくAIチャットボットを作成してみました。tool-calling loopsの仕組みはアイデア次第で色々作れるなと思いましたので、今回の知見を生かして今後も色々試してみたいです。
皆さんもぜひLangGraphを使って遊んでみてください！
（あとついでにアニメも観てください）
参考