DXC Data&AIPublicationへの投稿
🐵

MCP（Model Context Protocol）を実運用へ繋ぐ：OAuth2/PKCE認証と動的スキル拡張の実装

stkz
に公開
Model Context Protocol
llm
tech

この記事では、MCPの実装に取り組む中で見えてきた実運用での課題に対して、解決につながりそうな実装案を試してみた内容を紹介します。自分の取り組みを整理し言語化することで以後の取り組みに生かしていくと共に、MCPの実装方法を考える方の参考になれば幸いです。

1. MCPへの期待と、運用課題

2025年、AI界隈で話題になったMCP（Model Context Protocol）。LLMと外部ツールを標準的なプロトコルで繋ぐというこの技術に、私も大きな期待を持って触れ始めました。

私がMCPに期待を抱いた理由は、これまでのLLMワークフローにおける「コンテキスト管理」の難しさを痛感していたからです。前段の処理結果を後段のコンテキストに含める際、情報の質や量が少し変わるだけで最終回答の精度が劇的に変化してしまいます。だからこそ、必要な情報を適切かつスマートに提供できそうなMCPの仕組みは、非常に魅力的なものに映りました。

しかし、今回初めての実装に挑戦する中で、ただ繋げて動かすだけでは解決できない、運用を意識した際のいくつかの課題が見えてきました。

「手出しできない」運用中のMCPサーバー

運用が開始されたシステムにおいて、MCPサーバーは必ずしも自作のものとは限りません。たとえ自作であっても、基幹データに紐づくサーバーを、新しい分析ニーズが出るたびに安易に修正・デプロイし続けることは現実的ではありません。

エージェント本体の「硬直化」

同様に、一度組み上げられたエージェント側のロジックも、安定稼働を優先すれば頻繁な変更は避けたいものです。しかし、ユーザーからは「このデータをもっと加工してほしい」「統計的に処理して見せてほしい」という要望が次々と生まれます。

「ツール過多」による知能の低下

便利なツール（MCP）を増やせば増やすほど、LLMのコンテキストは膨れ上がり、選択の迷いや精度低下、そしてトークンコストの増大を招きます。

「サーバーもエージェントもいじれない。しかし、機能は柔軟に追加したい。しかも、LLMの賢さを損いたくない。」

この矛盾を解決するために、**動的な付加機能提示（スキルスロット）**というアプローチを思いつきました。本記事では、LangChainとFastMCPを用いた実践的な実装例をベースに、OAuth2/PKCEによる認証から、運用負荷を最小限に抑えるためのプラグイン的な拡張手法までを詳しく解説します。

2. 最小構成のMCP：まずは繋いでみる

MCP実装の第一歩として、まずは最も手軽な stdio（標準入出力）方式 でサーバーを構築し、エージェントから呼び出す最小構成を試してみます。

2.1 stdio接続サーバー (FastMCP)

PythonでMCPサーバーを構築する場合、fastmcp ライブラリを利用するのが最短ルートです。少ない行数で、MCPサーバーの機能をツールとしてLLMへ公開できます。

stdio接続のMCPイメージ

stdio接続のMCPサーバーコード概要

実装試行用のダミー機能として、特定利用者の特定期間の健康情報データを出力する機能を定義しました。

import json
import time
import pandas as pd
from fastmcp import FastMCP

mcp = FastMCP("HealthDataDB")  # FastMCPインスタンス
df = pd.read_csv("./sample_health_data.csv")

@mcp.tool()  # 関数をMCP機能化
def select_stdio(user_id: str, period_start: str, period_end: str) -> str:
    # DocstringsがLLMに呼んでもらうMCPの機能説明である。
    """
    Retrieve records from a database table by executing a filtered SELECT query.

    ``` sql
    SELECT *
    FROM table
    WHERE user_id = :user_id
    AND checkup_date >= :period_start
    AND checkup_date <= :period_end
    ```
    Arguments:
    - user_id (string, required)
    - period_start (str, required)
        - format: f"{year}-{month}-{day}"
        - example: "2023-04-03"
    - period_end (str, required)
        - format: f"{year}-{month}-{day}"
        - example: "2023-04-03"

    Return Value:
    - records (array<object>)
    """
    # 1. dfを、引数で指定された条件でフィルタリングする
    # 2. フィルタリングしたdfの内容を各行毎にdictに詰め、配列化する
    # 3. json.dumpsでテキスト化してreturnする
    return result

if __name__ == "__main__":
    mcp.run(transport="stdio")  # 通信プロトコルにstdioを指定
  • この方式の魅力は、
    • ローカルのスクリプトを、簡単にAIが指定するツールにできる点にあります。
    • MCP機能とエージェントフローの開発を分業できることはチーム開発において大きなメリットです。
  • 一方で、エージェントに取り込む外部データ生成のロジックが簡単で、エージェントと外部データ生成の機能を1人で開発するような場合は、エージェントコードと紐づけて実装するツール関数にすれば良いので、わざわざMCPにする必要はありません。streamable http形式の機能開発用と割り切ってもよいと思います。

2.2 MCPクライアント (LangChain)

次に、このサーバーを呼び出すクライアント側を構築します。今回は豊富なエコシステムで実装を簡易化できることを期待して LangChain を利用しました。

stdio接続のMCPクライアントコード概要

import asyncio
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

async def main():
    # 1. MCPクライアントの設定（stdio接続）
    # エージェントが子プロセスとしてMCPサーバーを起動する定義
    client = MultiServerMCPClient(
        {
            "mcp_server_stdio": {
                "transport": "stdio",
                "command": "python",
                "args": ["./mcp_server_stdio.py"],
            },
        }
    )

    # 2. MCPサーバーから利用可能なツール一覧を取得
    tools = await asyncio.wait_for(client.get_tools(), timeout=5.0)

    # 3. LLMの準備
    llm = ChatOpenAI(
        略記：モデル、URL、APIKey
    )

    # 4. LangChainエージェントの作成
    # 取得したMCPツールを直接エージェントに渡す
    agent = create_agent(llm, tools)

    # 5. 実行
    prompt = "Retrieve the health data and display only the statistical information of user: 12345 and period: 2023-06-01 ~ 2023-12-31 with markdown format."
    
    response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": prompt}]}
    )

    # 最終的な回答を表示
    print(response['messages'][-1].content)

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

単純に、LLMの実行結果を全て履歴に含めるとツールの実行指示の内容もコンテキストに含まれますが、LangChainの標準的なツール呼び出しフロー（create_agent -> agent.ainvoke）では、ツール実行のメッセージは履歴に含まれず、最終回答のコンテキストに入りません。これにより、最終回答生成の本質的な参考にならないツール実行情報がノイズになることを防ぐことができています。この実行履歴に含める内容をさらに制御したいならば、少なくとも当該処理部分は自分で実装する必要があります。

2.3 最低限実装で見えてくる課題

最小構成で動いた後、すぐにいくつかの実用上の制約が見えてきました。

  • 接続環境の制約: stdio方式はJupyter Notebookのような対話型環境からの接続が難しい場合があります。開発の初期段階で、少しずつ動作確認をしながら開発を進めたいときの制約になります。
  • セキュリティの不在: サーバー側から見ると、誰がツールを叩いているのかを識別する手段がなく、機密データを扱うには無防備です。
  • クライアント情報の欠如: 認証トークンなどを送る標準的な仕組みがstdio方式では確立しづらく、商用利用やマルチユーザー環境には不向きです。

3. セキュリティの仕掛け：OAuth2/PKCEの実装

stdio接続の限界を踏まえ、次は streamable http接続 への移行に挑戦します。HTTPで公開する場合、外部からツールが呼び出し放題になるリスクがあるため、今回は OAuth2 と、よりセキュアな PKCE (Proof Key for Code Exchange) を実装しました。

3.1 なぜPKCEが必要なのか

通常のOAuth2でも認可は可能ですが、クライアント側に秘密鍵を持たせにくい環境では、認可コードの横取り攻撃が懸念されます。PKCEは、実行時に動的に生成する「コード検証値（Verifier）」を用いることで、万が一認可コードが盗まれてもトークンを取得できないようにする仕組みです。

3.2 認可サーバーとPKCEの実装

まずは、認可コードの発行と検証を行う簡易的な認可サーバーを構築します。

PKCEの検証とトークン発行コード（抜粋）

認可サーバー側に以下を実装しました。

def verify_pkce(code_verifier: str, code_challenge: str) -> bool:
    # クライアントから送られたverifierをハッシュ化し、challengeと比較する
    hashed = hashlib.sha256(code_verifier.encode()).digest()
    challenge2 = base64.urlsafe_b64encode(hashed).rstrip(b"=").decode()
    return challenge2 == code_challenge

def create_jwt_token(user_info: dict, roles: list[str]):
    # 認証成功後、有効期限付きのJWTトークンを発行する
    expire = datetime.now(timezone.utc) + timedelta(minutes=60)
    payload = {**user_info, "roles": roles, "exp": expire.isoformat()}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

3.3 MCPサーバー側のトークン検証

MCPサーバー（FastMCP）側では、リクエストに含まれるBearerトークンが正当なものか、認可サーバーに問い合わせて検証する仕組みを組み込みました。

3.4 クライアントによる認可フローの自動化

クライアント（エージェントフロー）は、ツール実行前にバックグラウンドでPKCEのコード生成、認可コードの取得、トークン交換を自動で行います。

認可フロー図

これにより、正当な権限を持つエージェントだけが、特定のMCPツールを呼び出せるという、運用レベルに耐えうるセキュリティ基盤が整いました。

4. 運用課題への対処案：動的スキル拡張システム（スキルスロット）

セキュアな通信路が確保できても、実運用では、一度稼働したMCPサーバーやエージェントのロジックは簡単には変更できない、という保守性の制約に直面します。そこで私は、MCPツールの出力を利用側の自由な発想で加工できるプラグイン機構 を考案しました。

4.1 外部定義ファイル skills.json による拡張

この仕組みの肝は、エージェント本体のコードの外側に置かれた skills.json という定義ファイルです。

skills.json

{
    "select_http": "./skills/skills_select_http.py"
}

特定のMCPツール（ここでは select_http）が実行された際、このファイルに定義があれば、関連する付加機能（スキル）を動的にロードします。

4.2 動的ロードとプレースホルダによる効率化

エージェントの実装（org_create_agent）では、MCPツールの実行結果を受け取った直後、LLMに対して、このデータを加工するための追加ツール（スキル）を提示します。

動的インポートとプロンプトへのスキル注入ロジック（抜粋）

# skills.jsonに基づき、外部モジュールからツールを動的に読み込む
spec = importlib.util.spec_from_file_location(module_name, skill_file)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
outer_module_tools = [convert_to_openai_tool(tool) for tool in module.tools]

# LLMに対し、直前の結果をプレースホルダ {{PREV_RESULT}} で扱うよう指示
instruction_message = SystemMessage(content='...To process this data with the next tool, use "{{PREV_RESULT}}" as a placeholder...')
model_response = await self.llm.ainvoke(self.messages + [instruction_message], tools=outer_module_tools)

このスキルスロット方式の最大の利点は、LLMの推論コストの最適化です。最初から全てのツールをLLMに提示するのではなく、データ取得の必要性を選択したタイミングで初めて付加的なデータ加工スキルをコンテキストに注入します。これにより、最初の思考フェーズでのノイズを減らし、かつ加工フェーズでは直前のデータを {{PREV_RESULT}} というシンボルで扱うよう指示することで、LLMが長大なデータを引数として再生成する手間とミスを省いています。MCPサーバー側は一切変更せず、./skills/ ディレクトリにPythonファイルを追加するだけで新しいロジックを追加できるのです。

5. まとめ・所感

本稿では、MCPのポテンシャルを引き出すために、セキュアな認証基盤と、運用の柔軟性を両立させる動的スキル拡張の仕組みを実装しました。

5.1 構築したアーキテクチャの振り返り

今回実装した仕組みは、単なるツールの接続に留まらない、以下の3つの価値を提供しています。

  • 多層防御の実現: OAuth2とPKCEを用いることで、なりすましや不正利用を防止する強固な認可フローを構築しました。
  • 動的スロットによる拡張性: skills.json という定義ファイルを介在させることで、既存システムを無改造のまま、現場のニーズに応じてデータ加工ロジックを追加できる柔軟性を確保しました。
  • LLMへのノイズを抑える最適化: 必要なタイミングで必要なスキルだけを注入する手法は、大規模なツールセットを扱う際の精度維持に非常に有効です。

5.2 AIとの共同開発

今回、開発プロセスの大部分をAIサービスに問い合わせながら進めました。コードの生成や複雑なシーケンスの実装においてAIに下書きさせ、LangChainの独自仕様に起因するデバッグにおいても優秀なパートナーとなりました。AIが生成したコードの「良し悪し」を判断し、自分の設計思想に合わせて修正を加える過程で、自分自身の技術的な理解も一層深まったと感じています。

5.3 おわりに

MCPのような外部知識の導入手法は発展途上であり、後続の手法も登場してきています。動かしながら理解が深まったので、今後は新しい手法をより早く試していこうと思います。今回の私の試行が誰かの参考になれば幸いです。

DXC Data&AI
DXC Data&AIPublication

Discussion