AIエージェント構築＆運用 Advent Calendar 2025

FastMCPで始めるAIエージェント開発 - LangChainと独自MCPサーバを連携させてみた

Last updated at 2025-12-07Posted at 2025-12-07

はじめに

先日とあるコンテストに出場したが、その中でMCPサーバを活用したAIエージェント開発を行なった。
その際、MCPついてはQiitaなどを読んで理解していたつもりだったが、実際には活用したことがないことに気づいた。
なので今回は簡易的にMCPサーバを作成し、簡易的なLLMアプリケーションから呼び出して動作確認をしてみることにした。

対象者

MCP関連の実装をしたことがない方・慣れていない方
LangChainを利用して開発したことがある方

前提条件

MCPとはを理解していること
OpenAIのAPIトークンを取得済みであること
LangChainのインストールができていること
mcpなどの必要なパッケージがインストールできていること

MCPサーバの開発

まずは簡易的にMCPサーバを作成する。
今回はLLMアプリケーションから私moshimoの情報が取得できるように、moshimo専用のMCPサーバを作成することにした。

MCPサーバを作成するためにFastMCPというライブラリを利用する。
FastMCPはMCPアプリケーションを簡単に構築できるライブラリである。
中身の詳細や導入手順は下記のGithubを参照してください。

まず、FastMCPクラスを宣言する。

from fastmcp import FastMCP

mcp = FastMCP(name="moshimo-mcp-server")

次に、@mcp.toolを利用し、mcpサーバとして利用できるツールを作成する。
今回はmoshimoのプロフィールを取得するツールとスケジュールを取得するツールの二つを作成する。
簡易的な実装のため固定値をreturnしているだけだが、本来は外部システムと連携し情報を取得するのが一般的である。

@mcp.tool()
def moshimo_profile() -> dict:
    """moshimoのプロフィールを取得"""
    return {
        "name": "moshimogood",
        "age": 29,
        "github": "https://github.com/moshimogood",
        "twitter": "https://x.com/jibunwomigakuu"
    }

@mcp.tool()
def moshimo_schedule() -> dict:
    """moshimoのスケジュールを取得"""
    return {
        "2025-12-06": "休暇",
        "2025-12-07": "Qiita記事執筆",
        "2025-12-08": "仕事"
    }

エントリーポイントにてmcp.run()を実行することで、サーバを起動している。

if __name__ == "__main__":
    mcp.run()

LLMアプリケーションの開発

MCPからLangchainへの変換用関数の作成

LangChainにてMCPのツールを利用したいが、互換性がないためLangChainツール用に変換する必要がある。
MCPツールは様々なシステムから利用できるように標準化されたインターフェースを持っている。
具体的にはMCPツールはJSON Schemaで定義されているが、LangChainはPydanticベースの型付きPythonオブジェクトを期待するため変換が必要となる。

下記に実装を示す。
mcp_toolsを受け取り、それをLangChainが理解できるStructuredToolオブジェクトに変換する処理を書いている。
具体的にはStructuredTool.from_functionを利用し、変換用関数を渡している。
注目点としては以下の2つある。

非同期関数(coroutine)を利用する
- MCPクライアントのsession.call_tool()は非同期関数であるため、それを呼び出すラッパー関数もasync defで定義し、StructuredTool.from_function()のcoroutineパラメータに渡す必要がある
クロージャー関数を利用する
- pythonのループ変数は全イテレーションで変数が共有されるため、各ツールごとに独立したスコープを作る必要がある。今回はmake_caller関数のようなファクトリ関数を作成した

def create_langchain_tools(session: ClientSession, mcp_tools) -> List[StructuredTool]:
    """MCPツールをLangChainツールに変換"""
    langchain_tools = []
    for tool in mcp_tools:
        def make_caller(tool_name):
            async def call_mcp_tool(**kwargs):
                result = await session.call_tool(tool_name, kwargs)
                return str(result.content[0].text) if result.content else ""
            return call_mcp_tool
        
        langchain_tool = StructuredTool.from_function(
            coroutine=make_caller(tool.name),
            name=tool.name,
            description=tool.description
        )
        langchain_tools.append(langchain_tool)
    
    return langchain_tools

クエリ実行用関数の作成

次に、クエリを実行する関数を作成する。
処理フローは以下の通り。

ユーザからのクエリを利用し、システムプロンプトとユーザプロンプトを作成し、llm_with_toolsを実行する
レスポンスからツールを利用しない場合は、そのまま最終回答を作成する
レスポンスからツールを利用する場合はツールを利用する（複数ある場合も、順に全て利用）
ツールからの結果をメッセージに追加し、最終回答を作成する

async def run_query(llm_with_tools, langchain_tools: List[StructuredTool], query: str):
    """クエリを実行してツール呼び出しと結果表示を行う"""
    messages = [
        {"role": "system", "content": "あなたは親切なアシスタントです。利用可能なツールを使って質問に答えてください。"},
        {"role": "user", "content": query}
    ]
    
    response = await llm_with_tools.ainvoke(messages)

    # ツール呼び出しがある場合は実行
    if response.tool_calls:
        # アシスタントのメッセージを追加
        messages.append({"role": "assistant", "content": response.content or "", "tool_calls": response.tool_calls})
        
        for tool_call in response.tool_calls:
            tool = next((t for t in langchain_tools if t.name == tool_call["name"]), None)
            if tool:
                result = await tool.ainvoke(tool_call["args"])
                print(f"{tool_call['name']}: {result}")
                
                # ツール実行結果をメッセージに追加
                messages.append({
                    "role": "tool",
                    "content": str(result),
                    "tool_call_id": tool_call["id"]
                })
        
        # ツール結果を元に最終回答を生成
        final_response = await llm_with_tools.ainvoke(messages)
        print(f"\n最終回答: {final_response.content if final_response.content else '(内容なし)'}")

    else:
        # ツールを使わない場合はそのまま表示
        print(f"\n最終回答: {response.content}")

メイン関数

最後にメイン関数を作成する。
処理フローは以下の通り。

StdioServerParametersを利用し、最初に作成しておいたmcpサーバmoshimo_mcp_server.pyを起動する
MCPセッションを開始
MCPサーバからツール一覧を取得し、LangChainツールに変換
LLMを作成し、ツールをバインド
質問を実行（今回は時間の関係上固定値で質問を用意している）

async def main():
    server_params = StdioServerParameters(
        command=sys.executable,
        args=["./moshimo_mcp_server.py"],
        env=dict(os.environ)
    )

    # MCPセッションを維持したまま実行
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # ツール一覧を取得してLangChainツールに変換
            tools_result = await session.list_tools()
            langchain_tools = create_langchain_tools(session, tools_result.tools)
            
            # LLM設定
            llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2, openai_api_key=os.getenv("OPENAI_API_KEY"))
            llm_with_tools = llm.bind_tools(langchain_tools)
            
            # 質問を実行
            await run_query(llm_with_tools, langchain_tools, "moshimoのプロフィールとスケジュールを教えて")

if __name__ == "__main__":
    asyncio.run(main())

動作確認

demo用のpythonを実行する。
※uvで実行しているが、ここは各自の環境に応じて適宜変更してください

uv run langchain-demo.py

moshimoのプロフィールとスケジュールを教えてという質問に対して以下の情報が返ってきた。
ちゃんとmoshimoのプロフィールとスケジュールを返してくれています。

逆にmoshimoとは関係のない質問をしてみる。
pythonの良いところを簡単に教えてという質問にした。
moshimoの情報は入っていないため期待通り。

最後に

MCPサーバをLangChainで利用してみたが、簡単に実装できる一方で学べることも多かった。
このような検証ができたことはコンテストに出場したおかげだと思うのでやはり挑戦することには価値がある。
技術は自分で触って確かめることが最重要だと思うので、これからも続けたい。

参考コード

moshimo_mcp_server.py

from fastmcp import FastMCP

mcp = FastMCP(name="moshimo-mcp-server")

@mcp.tool()
def moshimo_profile() -> dict:
    """moshimoのプロフィールを取得"""
    return {
        "name": "moshimogood",
        "age": 29,
        "github": "https://github.com/moshimogood",
        "twitter": "https://x.com/jibunwomigakuu"
    }

@mcp.tool()
def moshimo_schedule() -> dict:
    """moshimoのスケジュールを取得"""
    return {
        "2025-12-06": "休暇",
        "2025-12-07": "Qiita記事執筆",
        "2025-12-08": "仕事"
    }

if __name__ == "__main__":
    mcp.run()

langchain-demo.py

import asyncio
import os
import sys
from typing import List
from langchain_core.tools import StructuredTool
from langchain_openai import ChatOpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from dotenv import load_dotenv

load_dotenv()


def create_langchain_tools(session: ClientSession, mcp_tools) -> List[StructuredTool]:
    """MCPツールをLangChainツールに変換"""
    langchain_tools = []
    for tool in mcp_tools:
        def make_caller(tool_name):
            async def call_mcp_tool(**kwargs):
                result = await session.call_tool(tool_name, kwargs)
                return str(result.content[0].text) if result.content else ""
            return call_mcp_tool
        
        langchain_tool = StructuredTool.from_function(
            coroutine=make_caller(tool.name),
            name=tool.name,
            description=tool.description
        )
        langchain_tools.append(langchain_tool)
    
    return langchain_tools


async def run_query(llm_with_tools, langchain_tools: List[StructuredTool], query: str):
    """クエリを実行してツール呼び出しと結果表示を行う"""
    messages = [
        {"role": "system", "content": "あなたは親切なアシスタントです。利用可能なツールを使って質問に答えてください。"},
        {"role": "user", "content": query}
    ]
    
    response = await llm_with_tools.ainvoke(messages)
    
    # ツール呼び出しがある場合は実行
    if response.tool_calls:
        # アシスタントのメッセージを追加
        messages.append({"role": "assistant", "content": response.content or "", "tool_calls": response.tool_calls})
        
        for tool_call in response.tool_calls:
            tool = next((t for t in langchain_tools if t.name == tool_call["name"]), None)
            if tool:
                result = await tool.ainvoke(tool_call["args"])
                print(f"{tool_call['name']}: {result}")
                
                # ツール実行結果をメッセージに追加
                messages.append({
                    "role": "tool",
                    "content": str(result),
                    "tool_call_id": tool_call["id"]
                })
        
        # ツール結果を元に最終回答を生成
        final_response = await llm_with_tools.ainvoke(messages)
        print(f"\n最終回答: {final_response.content if final_response.content else '(内容なし)'}")

    else:
        # ツールを使わない場合はそのまま表示
        print(f"\n最終回答: {response.content}")


async def main():
    server_params = StdioServerParameters(
        command=sys.executable,
        args=["./moshimo_mcp_server.py"],
        env=dict(os.environ)
    )

    # MCPセッションを維持したまま実行
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # ツール一覧を取得してLangChainツールに変換
            tools_result = await session.list_tools()
            langchain_tools = create_langchain_tools(session, tools_result.tools)
            
            # LLM設定
            llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2, openai_api_key=os.getenv("OPENAI_API_KEY"))
            llm_with_tools = llm.bind_tools(langchain_tools)
            
            # 質問を実行
            await run_query(llm_with_tools, langchain_tools, "pythonの良いところを簡単に教えて")

if __name__ == "__main__":
    asyncio.run(main())

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up