MCPサーバーを自作した話

はじめに

こんにちは。社会人2年目、AIエンジニア見習いのばんそこです。
最近はストーリー生成AIエージェントの開発を頑張っています。
AI(LLM)は面白いストーリーを生成してくれますが、記憶が儚く、すぐに忘れてしまいます。

生成したストーリーを守るために、生成したストーリーのファイルをローカルに保存したり、保存しているストーリーの情報を取得する仕組みが必要です。

ということで、ストーリー管理用のMCPサーバーを作って、LLMがストーリーファイルを扱えるようにしよう！というのがこの記事の目的になります。

環境・システム構成

技術スタック	機能/説明
Docker	Webアプリ、API(FastAPI) 、DB、Prismaの他に、MCPサーバー用のコンテナを作成します。
FastAPI	ユーザーからのリクエストを受け付け、LLMを実行します。 LLMがユーザーの意図を解釈し、ツールを使用するか、どのツールを使用するかを判断します。
FastMCP	ファイル操作などのツールが実装されているサーバーです。 LLMから依頼を受けて、ツールを実行し、結果を返却します。

実装内容

LLMとMCPサーバーを接続してツールを使えるようにするまでの流れを説明していきます。

環境構築

Dockerを用いているため、MCPサーバー用のコンテナを定義し、FastAPIなどの他のサービスと連携できるように設定します。

mcp_server/Dockerfile

FROM python:3.11-slim

WORKDIR /app

RUN apt-get update && apt-get install -y pipx
RUN pipx install poetry==2.1.3

ENV PATH="/root/.local/bin:$PATH"
COPY pyproject.toml poetry.lock ./
RUN poetry install --no-root --no-directory

COPY . .

EXPOSE 8001

CMD ["sh"]

Pythonのバージョン管理にpoetryを使っているため、poetry runで起動します。

docker-compose.yml

mcp_server:
  build:
    context: ./mcp_server
  ports:
    - "8001:8001"
  volumes:
    - ./mcp_server:/app
  networks:
    - my-network
  command: poetry run fastmcp run main.py:mcp --transport http --host 0.0.0.0 --port 8001

mcp_server/およびbackend/のディレクトリ構成は以下の通りです。

mcp_server/のディレクトリ構成

FastMCPで構築されたツール実行用のサーバーです。
FastMCPは公式ドキュメントを参考にプロジェクトにインストールします。

mcp_server/
├── main.py                    # MCPサーバーのエントリーポイント
├── pyproject.toml             # 依存関係ファイル
├── poetry.lock
├── Dockerfile
├── stories/                 # ストーリーデータ保存ディレクトリ
│  └── xxxxx.txt …
└── tools/                   # MCPツール実装ディレクトリ
    ├── story_tools.py       # ツール統合クラス
    └── create_story.py      # ストーリー作成ツール（他:読み取り、一覧表示、削除…）

backend/のディレクトリ構成

FastAPIアプリケーションのディレクトリです。LLMエージェントを実行し、ツール使用時はMCPサーバーとの通信を行います。

backend/
├── main.py                    # FastAPIアプリケーションのエントリーポイント
├── mcp_client.py             # MCPサーバーとの通信クライアント
├── pyproject.toml
├── poetry.lock
├── Dockerfile
├── routers/                # FastAPIルーターディレクトリ
│   ├── chat.py             # チャット機能（MCP連携の中核）
│   ├── prompt.md           # AIエージェント用システムプロンプト
│   └── loadPrompt.py       # プロンプト読み込みユーティリティ
└── tools/                  # ツール定義ディレクトリ
    └── story_tools.py      # MCPツールのLangChain用定義

MCPサーバーとLLMの接続

まずは、LLMがMCPサーバーにツール実行を依頼できるように準備をします。

curlで色々試していたのですが、
はじめはなかなかMCPサーバーとLLMの接続ができず、Geminiとともに激闘した末に、
「コミュニケーションは、どこまでも丁寧に。」 という結論に至りました。

MCPサーバーは、例えるなら手続きが厳格な市役所の窓口。正しい手順を踏まないと、相手にしてもらえません。

MCPサーバーに接続する側の処理から作りながら、MCPサーバーの動きを追ってみます。

整理券の発行（セッション初期化）

まずは、総合受付で「どのようなご用件ですか？」と聞かれ、整理券を受け取るところから始まります。整理券（session_id）がなければ、そもそも待合室にすら入れません。

backend/mcp_client.py

# 総合受付にて
init_response = requests.post(
    MCP_URL,
    json={
        "jsonrpc": "2.0", "method": "initialize",
        "params": {
            # ...
            "clientInfo": { "name": "backend-agent" } # ← 身分証明書です
        },
        # ...
    }
)
# サーバーから「あなたの番号はこちらです」と整理券（セッションID）を受け取る
session_id = init_response.headers.get("mcp-session-id")

待合室での準備完了（初期化完了通知）

整理券を手に入れたら、待合室で自分の番号が呼ばれるのを待ちます。そして、「こちらの準備はできましたよ」と窓口に合図を送ります。

backend/mcp_client.py

# 待合室にて
requests.post(
    MCP_URL,
    headers={"mcp-session-id": session_id}, # ← 整理券を提示
    json={
        "jsonrpc": "2.0",
        "method": "notifications/initialized", # ←「準備OKです」の合図
        "params": {}
    }
)

窓口での正式な依頼（ツール呼び出し）

最後に、手続き用の整理券を持って、担当窓口に行きます。
「〇〇というツール（手続き）をお願いします」と実際に依頼をすることができます。

backend/mcp_client.py

# 担当窓口にて
tool_response = requests.post(
    MCP_URL,
    headers={"mcp-session-id": session_id}, # ← 整理券を再度提示
    json={
        "jsonrpc": "2.0", "method": "tools/call",
        "params": {
            "name": tool_name,       # ←「住民票発行ツール」をお願いします
            "arguments": arguments   # ← 必要な書類（引数）はこちらです
        },
        # ...
    }
)

ツールの使い方

LLMがMCPツールを呼び出す準備が整ったため、いよいよ実際にLLMが使うツールを実装します。
ここでは、LLMが生成したストーリーをファイルに書き込んで保存するcreate_storyツールを実装します。

使えるツールをMCPサーバー側でまとめて定義

mcp_serverディレクトリ内に、ツールをまとめて定義するファイルを作成します。mcp_server/main.pyで直接ツールの実装内容を書くこともできますが、拡張性のためファイルを分離します。

mcp_server/tools/story_tools.py

from fastmcp import FastMCP
from .create_story import create_story as create_story_impl

class StoryTools:
    def __init__(self, mcp: FastMCP, stories_dir: str):
        self.mcp = mcp
        self.stories_dir = stories_dir
        self.register_tools()
    
    def register_tools(self):
        """
        全ツールを登録
        """
        self.mcp.tool(self.create_story)
    
    def create_story(self, title: str, content: str) -> dict:
        """
        作成したストーリーをテキストファイルとして保存する
        """
        return create_story_impl(title, content, self.stories_dir)

FastAPIはbackend内のディレクトリしか見ないので、LLM用のツール定義ファイルが別に必要です。
上記mcp_server内のMCPツール定義ファイルから、各ツールの定義部分を抜き出したファイルを作成します。
（割愛しますが、mcp_server/tools/story_tools.pyの内容からbackend側に自動でファイルを生成することもできます。）

backend/tools/story_tools.py

from langchain_core.tools import tool

@tool
def create_story(title: str, content: str) -> dict:
    """作成したストーリーをテキストファイルとして保存する"""
    pass

tools = [create_story]

実際のツール（create）の処理内容

LLMが生成したストーリーをファイルに書き込んで保存するcreate_storyツールを作成します。

mcp_server/tools/create_story.py

import logging
import os

def create_story(title: str, content: str, stories_dir: str) -> dict:
    """
    作成したストーリーをテキストファイルとして保存する
    """
    logging.info(f"ストーリー作成")

    filename = f"{title}.txt"
    filepath = os.path.join(stories_dir, filename)

    if os.path.exists(filepath):
        logging.warning(f"ストーリー '{title}' は既に存在します。")
        return {"error": "Story with this title already exists"}
    
    try:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)
        
        logging.info(f"ストーリーを保存しました：{filepath}")
        return {"message": "Story created successfully", "title": title}
    
    except Exception as e:
        logging.error(f"ストーリー保存中にエラーが発生しました: {e}")
        return {"error": f"Failed to save story file: {e}"} 

LLMのツール呼び出し処理

ツールができたので、backend側でLLMがcreate_storyを呼び出せるように、処理を追加します。

エージェントの定義

llm.bind_tools(tools)は、Pythonの関数定義をLLMが解釈可能な形式（OpenAI ToolsのJSON Schema）に変換し、モデルに紐付ける重要な処理です。これにより、LLMは各ツールの機能、引数、説明を理解し、思考の選択肢として扱えるようになります。

backend/routers/chat.py

from tools.story_tools import tools
from .loadPrompt import load_prompt_from_file

# LLMにツールを認識させ、プロンプトやパーサーと連結してエージェントを作成
llm_with_tools = llm.bind_tools(tools)

# プロンプトテンプレート作成・読み込み
prompt = ChatPromptTemplate.from_messages([
        ("system", load_prompt_from_file("prompt.md")),
        MessagesPlaceholder(variable_name="chat_history"),
        ("user", "{user_input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ])

agent = (
    preprocesser
    | prompt
    | llm_with_tools
    | OpenAIToolsAgentOutputParser()
)

ツールの呼び出しと応答生成の処理

1. エージェントの最初の判断を仰ぐ
   • agent.invokeを初回呼び出し。LLMはユーザーの入力と履歴から、ツールを使うべきか（Act）、直接応答すべきか（Finish）を判断します。
2. 実行
   • LLMがツール使用を判断すると、応答はlist形式（ToolAgentActionのリスト）で返されます。ここからツール名と引数を抽出し、call_mcp_toolでMCPサーバーに実行を依頼します。
3. 最終処理
   • ツールの実行結果をintermediate_stepsに格納し、再度agent.invokeを呼び出します。これが観測ステップです。LLMは「ツールを使ったらこの結果が得られた」という事実を元に、ユーザーへの最終的な応答を考え、AgentFinishとして返します。

backend/routers/chat.py

# 1. ユーザー入力をもとに、エージェントの最初の判断を仰ぐ
response = agent.invoke({
    "user_input": user_input.text,
    "chat_history": chat_history_items,
    "intermediate_steps": []
})

# 2. エージェントの判断に応じて処理を分岐
# パターンA: ツールを使わず、直接応答を返す場合
if isinstance(response, AgentFinish):
    ai_response_content = response.return_values['output']

# パターンB: ツールを使う指示が返ってきた場合
elif isinstance(response, list):
    response_action = response[0]
    tool_name = response_action.tool
    tool_args = response_action.tool_input
    
    # 3. 指示されたツールを実行
    mcp_result = call_mcp_tool(tool_name, tool_args)
    
    # 4. ツールの実行結果をエージェントにフィードバックし、最終的な応答を生成させる
    final_response = agent.invoke({
        "user_input": user_input.text,
        "chat_history": chat_history_items,
        "intermediate_steps": [(response_action, json.dumps(mcp_result))]
    })

    if isinstance(final_response, AgentFinish):
        ai_response_content = final_response.return_values['output']
    else:
        ai_response_content = "ツールの実行結果から最終的な応答を生成できませんでした。"

プロンプトによる制御

仕上げに、ユーザーがストーリーの作成を依頼した時にLLMがツールを呼び出せるようプロンプトで指示してあげることで、以下の流れでストーリーを作成・保存できるようになりました！

ユーザー: 「ストーリー作成して」
   ↓
LLM: 詳細確認（ジャンル・要素）
   ↓
ユーザー: 詳細回答
   ↓
LLM: ストーリー生成 → create_story ツール実行指示
   ↓
MCP: ファイル保存 → 結果返却
   ↓
LLM: 「保存しました」

まとめ

ストーリー管理用のMCPサーバーを作ったことで、LLMがストーリーファイルを扱い、安全安心なストーリー作成と保存ができるようになりました👏👏
色々苦労したこともありますが、MCPサーバーを自作した経験はプライスレス。
便利なのでみなさんも作ってみてください！