はじめに
「AI エージェント」を使おう、とよく聞きます。また、にわかに「MCP」が話題になっています。
まず「Function Calling」を試してみました。
続いて「MCP」が何か調べてみたいと思います。
MCP とは
MCP
は、Anthropi 社が公開した規格です。(2025 年)
モデルコンテキストプロトコル(MCP) - Anthropic
USB-C がデバイスを様々な周辺機器やアクセサリーに接続する標準的な方法を提供するように、MCP は AI モデルを異なるデータソースやツールに接続する標準的な方法を提供します。
よく分かりませんね。別の記事を探してみました。
[MCP再入門]「MCPはAIアプリにとってのUSB-C」がしっくりこなかったあなたに #Python - Qiita
Function Calling
は、アプリで用意した 機能/関数(ツール)
を、LLM に「呼出」させる機能でした。これによって LLM はデータソースやツールに接続できます。MCP
は、この実装をアプリの外部で実装できるようにする仕組のようです。
Function Calling とは
まず Function Calling
を試してみました。
LangChain を使って Function Call してみた #AI - Qiita
これを踏まえて MCP
を試していきます。
LangChain で MCP を使ってみる
MCP
も LangChain を使って実装できるようです。
LangChainとModel Context Protocol(MCP)の連携方法|生成AI活用研究部
これを試してみます。
実行環境を用意する
AI プログラムの実行環境は、高速な計算するために大きなメモリや GPU を使います。そこでこれまで Google Colab を使ってきました。
ところが、MCP プログラムは後述するように、Colab 環境で期待したように動きません。
高性能な実行環境をクラウドサービスで用意することもできますが、Gemini API のようなクラウドサービスを呼出するなら CPU のみの実行環境でいいでしょう。
ただし、Python プログラムが実行できるようにしておきます。
LangChain の MCP Adapter を使う
公式の Python 用 SDK を使います。必要なライブラリをインストールします。
$ pip install mcp
また、LangChain の MCP Adapter
を使います。
$ pip install langchain langchain_mcp langchain_mcp_adapters
MCP サーバを実装する
MCP サーバを実装します。
Function Calling
で 機能/関数
を用意したときの実装です。↓
from langchain.tools import tool
@tool
def add_numbers(x: int, y: int) -> int:
"""Add two numbers together."""
return x + y
(後略)
これを変更します。↓
import mcp
mcp = mcp.server.fastmcp.FastMCP("mcp-server-example")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
@mcp.tool()
def ask_information(word: str) -> str:
"""Get information about the given keyword."""
match word:
case "なごや個人開発者の集い":
answer = "毎週日曜日に開催する定例オフライン開発会です。 ソフト・ハードのエンジニアだけでなく、デザイナー、クリエイター、マーケターの方々が集い、もくもく作業するもよし、開発・制作の相談をするもよし、単に作品の自慢をするもよし、協力してプロダクトを作るもよし"
case _:
answer = "不明"
return answer
MCP サーバとして実行できるようにします。↓
(前略)
if __name__ == "__main__":
mcp.run(transport="stdio")
server.py
で保存しておきます。
MCP クライアントアプリを実装する
続いて、MCP クライアントアプリを実装します。
Function Calling
で 機能/関数の呼出
したときの実装です。↓
import langchain_google_genai
import langchain_core.prompts
import langchain.agents
# モデルを準備
model = langchain_google_genai.ChatGoogleGenerativeAI(
model="gemini-2.0-flash"
)
(中略)
# ツールを指定
tools = [add_numbers, ask_information]
# エージェントの初期化
agent = langchain.agents.create_tool_calling_agent(model, tools, prompt)
executor = langchain.agents.AgentExecutor(agent=agent, tools=tools)
(中略)
# 推論を実行
output = executor.invoke({"query": query})
print(output)
これを変更します。
# ツールを指定
tools = [add_numbers, ask_information]
この箇所が↓
# MCP サーバ呼出の設定
params = mcp.StdioServerParameters(
command="python",
args=["server.py"],
)
# MCP サーバを呼出
async with mcp.client.stdio.stdio_client(params) as (read, write):
async with mcp.ClientSession(read, write) as session:
await session.initialize()
tools = await langchain_mcp_adapters.tools.load_mcp_tools(session)
コマンド python server.py
が実行され、その実行セッションと接続して tools
オブジェクトをセットします。↑
server.py
を予め実行しておくと書かれた記事もあります。上記のコードはクライアントアプリが実行してくれます。
上記のコードは Gemini API サービスを使っています。予め API キーを取得して環境変数にセットするなどして下さい。
LangChain の Gemini統合 を試す|npaka
変更後のコードは↓
import langchain_google_genai
import langchain_core.prompts
import langchain.agents
import mcp
import langchain_mcp_adapters.tools
import asyncio
async def main():
# モデルを準備
model = langchain_google_genai.ChatGoogleGenerativeAI(
model="gemini-2.0-flash"
)
# プロンプトを準備
prompt = langchain_core.prompts.chat.ChatPromptTemplate.from_template(
"""Question: {input}
Thought: Let's think step by step.
Use one of registered tools to answer the question.
Answer: {agent_scratchpad}"""
)
# MCP サーバ呼出の設定
params = mcp.StdioServerParameters(
command="python",
args=["server.py"],
)
# MCP サーバを実行
async with mcp.client.stdio.stdio_client(params) as (read, write):
async with mcp.ClientSession(read, write) as session:
await session.initialize()
tools = await langchain_mcp_adapters.tools.load_mcp_tools(session)
# エージェントを用意
agent = langchain.agents.create_tool_calling_agent(model, tools, prompt)
executor = langchain.agents.AgentExecutor(agent=agent, tools=tools)
query = "なごや個人開発者の集いとは何ですか?"
# 推論を実行
output = await executor.ainvoke({"input": query})
print(output)
asyncio.run(main())
MCP サーバ+クライアントアプリを実行する
上記のコードを実行します。結果は↓
{'input': 'なごや個人開発者の集いとは何ですか?', 'output': '名古屋個人開発者の集いとは、毎週日曜日に開催される定例オフライン開発会です。 ソフト・ハードのエンジニアだけでなく、デザイナー、クリエイター、マーケターの方々が集い、もくもく作業するもよし、開発・制作の相談をするもよし、単に作品の自慢をするもよし、協力してプロダクトを作るもよし、という集まりです。'}
Google Colab で実行してみる
前述のコードを Google Colab で実行できないものでしょうか。試してみました。
そのために MCP サーバのコードと MCP クライアントアプリのコードを同時に実行しないといけないのですが、Google Colab ないし Jupyter Notebook は複数のセルを同時に実行できません。
そこで、server.py
のコードを一旦ファイルに保存して、クライアントアプリのコードで実行して貰うことにします。
%%file server.py
import mcp
mcp = mcp.server.fastmcp.FastMCP("mcp-server-example")
@mcp.tool()
(中略)
if __name__ == "__main__":
mcp.run(transport="stdio")
%%file
を実行することで、セルの内容をファイルに保存できます。↑
その上で MCP クライアントアプリのコードを実行すると↓
asyncio.run(main())
RuntimeError: asyncio.run() cannot be called from a running event loop
これを回避するために↓
import nest_asyncio
nest_asyncio.apply()
この上で改めて実行すると↓
# MCP サーバを実行
async with mcp.client.stdio.stdio_client(params) as (read, write):
UnsupportedOperation: fileno
クライアントアプリが MCP サーバを実行して、その実行セッションと接続します。このとき、通信は stdio(標準入出力)
が使われます。↓
mcp.run(transport="stdio")
mcp.client.stdio.stdio_client
が stdio(標準入出力)
にアクセスしようとしたとき、Google Colab がサポートしていなくて発生するようです。
これを解消できないか調べてみましたが分かりませんでした。
他の MCP サーバを試す
これまで自作した MCP サーバを試してみましたが、あらかじめ実装された MCP サーバが公開されています。
そのうち、以下の MCP サーバを試してみましょう。
servers/src/filesystem at main · modelcontextprotocol/servers
併せて、複数の MCP サーバをまとめて設定できるように、コードを変更します。↓
import langchain_google_genai
import langchain_core.prompts
import langchain.agents
import mcp
import langchain_mcp_adapters.client
import asyncio
async def main():
(中略)
# MCP サーバ呼出の設定
params = {
(中略)
"filesystem": {
"command": "npx",
"args": ["-y","@modelcontextprotocol/server-filesystem","~/"]
}
}
# MCP サーバを呼出
async with langchain_mcp_adapters.client.MultiServerMCPClient(params) as client:
(後略)
command
が npx
になっています。実行環境で Node.js プログラムが実行できるようにしておきます。
(前略)
# エージェントを用意
agent = langchain.agents.create_tool_calling_agent(model, tools, prompt)
executor = langchain.agents.AgentExecutor(agent=agent, tools=tools)
# エージェントを実行
query = "ディレクトリ ~/ にどんなファイルがありますか?"
output = await executor.ainvoke({"input": query})
print(output)
asyncio.run(main())
実行した結果↓
{'input': 'ディレクトリ ~/ にどんなファイルがありますか?', 'output': '/home/(中略) ディレクトリには、以下のファイルとディレクトリがあります:\n.bash_history\n.bash_logout\n.bash_profile.bashrc\n.cache\n.config\n.local\n.npm\n.profile\n.python_history\n.ssh\n.vscode\n.vscode-server\ncode\n'}
呼出された MCP サーバがファイルの操作できるようです。
MCP サーバをリモートサーバで実行する
これまで MCP サーバは、クライアントアプリと同じ環境で実行して、stdio(標準入出力)
で通信してきました。
MCP は、さらに HTTP の SSE(サーバ送信イベント)
でも通信できるようにしています。これも試してみましょう。
MCP サーバのコードを修正します。↓
(前略)
if __name__ == "__main__":
mcp.run(transport="sse")
これを予め実行しておきます。
MCP クライアントアプリのコードを修正します。↓
(前略)
# MCP サーバ呼出の設定
params = {
(中略)
"example": {
"url": "http://(MCP サーバを実行したホスト):8000/sse",
"transport": "sse",
}
}
(後略)
こうすれば、クライアントアプリと別のサーバ機で MCP サーバを実行できるようになります。
MCP とは
MCP
は、基本の概念は Functin Calling
と同じ。ただし Function Calling
は機能/関数を「アプリ内」で実装したが、MCP
は「アプリ外」に実装します。アプリと機能/関数の呼出の仕様(プロトコル)を決めたのが MCP(Model Context Protocol)
と言えるでしょうか。
機能/関数をアプリの外部に持ってきて、決められた仕様で呼出できるようにしました。このことで、機能/関数をアプリごとに実装しなくて済みます。また、既に実装されている機能/関数を呼出するだけで済むようになります。
Python で作るアプリが、Node.js で作られた機能/関数を利用できたりします。自作した機能/関数が自作のアプリだけでなく Claude Desktop などのアプリで利用できたりします。
機能/関数をリモートサーバに置いて呼出できるので、クラウドサービスに MCP サーバを用意することが増えて、様々な AI アプリから利用されることが期待されます。