既存MCPサーバーの「もう少しこうだったら」を少コストで解決する ― FastMCP Middleware実践

記事タイトルとURLをコピーする

はじめに

MCPの普及が進むにつれ、ある程度知名度のあるSaaSでは公式なMCPが提供されていることがかなり多くなりました。

しかし実際に使ってみると、以下のように利用感が自らのユースケースにマッチしきらないこと・特に権限制御の粒度に課題を感じることは少なくありません

  • ツールが多すぎてLLMのコンテキストを圧迫する
  • 書き込み系のツールをうっかり呼ばれると困る
  • 特定のリソースに対してだけ操作を許可したいが絞り込めない

本記事では、FastMCPのProxy機能とMiddleware機能を組み合わせることで、既存のMCPサーバーの入出力をきめ細かくかつ手軽に制御する手法を解説します。

今回のブログは社内向けの実装サンプルを示す狙いも兼ねているので、弊社でもよく使っているSaaSであるBacklogのMCPを例にとって説明していきたいと思います。

備考

先日のブログで姉妹版として出すとコメントしていたFastMCPのOAuth Proxy / OIDC Proxy機能とは別の話です。

タイミング・題材的に紛らわしくてすみません(実務の都合上、本ブログの方が優先度が高くなってしまいました)。

既存MCPサーバーが自社のワークロードにうまくマッチしない場合にどう対処するか

よくある困りごと

多くのMCPサーバーは、提供するツールの全量をそのままクライアントに公開します。

たとえばプロジェクト管理ツールのMCPサーバーであれば、課題の参照だけでなく、プロジェクトの作成・削除・設定変更といった強力な操作もツール一覧に含まれることがあります。

これが生成AIを組み込んだクライアントツール(Coding Agent等)の利用上、問題になる理由は主に3つあります。

  1. 意図しない操作のリスク:LLMは与えられたツールを「使えるもの」として認識します。削除系のツールが見えていれば、文脈次第では呼び出してしまう可能性があります
  2. コンテキストウィンドウの浪費:ツール定義はトークンを消費します。使わないツールが数十個も列挙されていると、本来の作業に使えるコンテキストが圧迫されます*1
  3. 環境ごとのポリシー適用が困難:「本番環境では参照のみ」「サンドボックスなら書き込みOK」といった条件分岐をうまく表現できないケースがあります

各ツールの公式MCPサーバーの機能だけでは1~3の課題を解決できないケースは割合多いものです

業務の割と根幹となるシステムに関するMCPについて、そのようなケースに当てはまった場合は、SaaS側・API側の機能強化を悠長に待てないケースも多々あろうかと思います。

さりとて、自分で一からMCPサーバーのプログラムを構築・メンテナンスするというのも「ちょっと違うよなぁ・・・」という方がほとんどでしょう。

Proxyという選択肢

このような問題に対するアプローチの一つとして、MCPサーバーの手前にProxyサーバーを差し込むアプローチが考えられます。

クライアント(Coding Agentなど)はProxyに接続し、Proxyが既存のMCPサーバーへリクエストを中継します。

この中継の過程で、ツール一覧のフィルタリング、リクエスト内容の検査・拒否、レスポンスの加工などを差し込むことができます。

プロキシの概念図

FastMCPにはこのProxy構成を簡潔に実装するための機能が組み込まれており、わずかなコードで既存MCPサーバーをラップできます。

また、後半で具体要領を説明しますが、個人がローカルで利用する前提のProxyもサクッと用意できるCoding AgentのProxy兼MCPサーバーとして気軽に組み込める こともポイントです。

FastMCP Proxyの基本

create_proxy で既存MCPサーバーをラップする

FastMCPの create_proxy 関数を使うと、既存のMCPサーバーをそのまま中継するProxyサーバーを作成できます。

from fastmcp import FastMCP
from fastmcp.server import create_proxy

# MCPサーバーの設定をそのまま渡してProxyを作成
config = {
    "mcpServers": {
        "backlog": {
            "command": "npx",
            "args": ["backlog-mcp-server"],
            "env": {
                "BACKLOG_DOMAIN": "yourspace.backlog.com",
                "BACKLOG_API_KEY": "your-api-key-here", # 実際のワークロードでは環境変数などを通じてセキュアに読み込むとよいです
                "ENABLE_TOOLSETS": "space,project,issue,wiki",
            },
        }
    }
}
backlog_proxy = create_proxy(config, name="Backlog")

create_proxy は、リモートMCPサーバーのURL・特定ファイルパス・MCPサーバーの設定辞書・既存の FastMCP インスタンスなど、多様な形式の接続先を受け付けます。

上記の例では、stdioトランスポートで起動するMCPサーバーの設定をそのまま辞書として渡しています。

この時点で backlog_proxy は元のMCPサーバーと同じツール・リソース・プロンプトをそのまま公開する透過的なProxyとして機能します。ここにMiddlewareを追加することで、入出力の制御を実現します。

Middlewareの概要

Middlewareとは

FastMCPのMiddlewareは、特定のタイミングでリクエストをインターセプトするための仕組みです*2

通信プロトコルの複雑な詳細を意識することなく、「ツールを実行する」「ツール一覧を取得する」といったアプリケーションとして意味のある単位で、直感的に処理を差し込めるように工夫しているのが特徴です。

Middleware クラスを継承し、必要なタイミング(フック)に対して必要な処理を定義・オーバーライドするだけで、自作Middlewareを実装できます。

from fastmcp.server.middleware import Middleware, MiddlewareContext

class MyMiddleware(Middleware):
    async def on_list_tools(self, context: MiddlewareContext, call_next):
        """ツール一覧の取得時に呼ばれる"""
        tools = await call_next(context)  # 元のサーバーからツール一覧を取得
        # ここでフィルタリングや加工が可能
        return tools

    async def on_call_tool(self, context: MiddlewareContext, call_next):
        """ツール実行時に呼ばれる"""
        # context.message.name でツール名を取得
        # context.message.arguments で引数を取得
        return await call_next(context)  # 元のサーバーにリクエストを転送

主なフック一覧

フック名 関連するタイミング 用途
on_call_tool ツール実行時 実行可否の判定、引数の書き換え
on_list_tools ツール一覧取得時 不要ツールの除外
on_read_resource リソース読み取り時 リソースアクセスの制御
on_get_prompt プロンプト取得時 プロンプトの加工
on_message 全メッセージ(高頻度) ログ記録、共通処理

call_next(context) を呼ぶことで処理をチェーンの次段(最終的には元のMCPサーバー)に転送し、呼ばなければリクエストをそこで止められます。

実装例

on_list_tools によるツール一覧のフィルタリング

MCPクライアントがサーバーに接続すると、まず tools/list でツール一覧を取得します。この応答をフィルタリングすることで、クライアントから見えるツールを制御できます。

以下は、実際のBacklog MCP Serverに対するフィルタリングの例です。

async def on_list_tools(self, context: MiddlewareContext, call_next):
    tools = await call_next(context)
    HIDDEN_PREFIXES = (
        "add_project",
        "update_project",
        "delete_project",
        "delete_issue",
        "watching_list_items",
        "watching_list_count",
        "watching",
        "custom_fields",
        "milestone",
        "milestone_list",
        "delete_version",
    )
    filtered = [t for t in tools if not t.name.lower().endswith(HIDDEN_PREFIXES)]
    return filtered

この例では、プロジェクトの作成・更新・削除、課題の削除、ウォッチリスト関連、カスタムフィールドやマイルストーンといった「自身の用途では不要、あるいは危険な操作」をツール一覧から除外しています*3

ポイントは、元のMCPサーバーのコードを一切変更していないことです。Proxy側のMiddlewareだけで公開ツールを制御できるため、MCPサーバーのバージョンアップにも影響を受けません。

on_call_tool によるポリシーベースの実行制御

ツール一覧のフィルタリングだけでは不十分なケースがあります。LLMが過去のコンテキストからツール名を記憶していて、一覧に無いツールを直接呼び出す可能性もゼロではありません。

また、同じツールでも引数の内容によって許可・拒否を分けたい 場合もあります。

こうした場合は on_call_tool フックでリクエスト内容を検査し、条件に合わなければ例外を投げてブロックします。

from fastmcp.exceptions import ToolError

HIGH_RISK_TOOL_PATTERNS = ("update", "add", "delete")
SANDBOX_PROJECT_IDS = (1111111111, 2222222222) # 実在しないProjectIdです

class BacklogPolicyMiddleware(Middleware):
    async def on_call_tool(self, context: MiddlewareContext, call_next):
        tool_name = context.message.name
        args = context.message.arguments or {}

        # 書き込み系ツールは、サンドボックスプロジェクトに対してのみ許可
        if tool_name.startswith(HIGH_RISK_TOOL_PATTERNS):
            if "projectId" not in args or args["projectId"] not in SANDBOX_PROJECT_IDS:
                raise ToolError(
                    f"Tool '{tool_name}' can only be called for sandbox projects."
                )

        return await call_next(context)

このMiddlewareは、updateadddelete で始まるツール名の呼び出しに対して、引数の projectId が書き込み系の処理を許容できるサンドボックスプロジェクトのIDリストに含まれているかを検査します*4

含まれていなければ ToolError を送出してリクエストを拒否します。

これにより、「参照系のツールは全プロジェクトで自由に使えるが、書き込み系のツールはサンドボックスプロジェクトでしか実行できない」というポリシーを宣言的に記述できます。

補論: ToolErrorMcpError の使い分け

FastMCPサーバーで異常を表現するにあたって主要なエラーには McpErrorToolError の2種類があります*5。 この2つは エラーが属するレイヤーが異なります。

MCPが採用するJSON-RPCには、エラーを表現するレイヤーが2つあります。

  • JSON-RPC層:メソッドの呼び出し自体が成功したか(result vs error
  • ツール結果層:ツールの処理結果が成功だったか(result 内の isError フラグ)

ToolErrorMcpError はこの2つのレイヤーに対応しています。

例外 エラーのレイヤー JSON-RPCでの表現
ToolError ツール結果層 result フィールド(isError: true
McpError JSON-RPC層 error フィールド

それぞれが生成するJSON-RPCレスポンスを見ると、違いがより明確になります。

ToolError の場合 ― JSON-RPCとしては正常応答(result)で、ツール結果層でエラーを表現:

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {"type": "text", "text": "Tool 'delete_issue' can only be called for sandbox projects."}
    ],
    "isError": true
  }
}

McpError の場合 ― JSON-RPC層でエラーを表現(result フィールドは存在しない):

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32601,
    "message": "Method not found: unknown_tool"
  }
}

「サンドボックスプロジェクト以外は拒否する」というのはツールのビジネスルールであり、プロトコルの異常ではありません。

ツールはリクエストを受け取り、内容を検査した上で拒否の判断を下しています。したがって今回の例では ToolError(=ツール結果層のエラー)でエラーを表現しています。

一方で McpError はクライアントのバージョン不整合や認証失敗など、ツール実行以前の問題に使うのが良いでしょう。

参考:FastMCP Middleware ドキュメント

その他、Proxyが活きるユースケース

Proxy機能のユースケース

FastMCP Proxyの活用は、ツールのフィルタリングや呼び出し拒否に留まりません。以下にいくつかのユースケースを紹介します。

入力のサニタイズ・書き換え

ツール呼び出しの引数を加工してから元のサーバーに転送できます。

たとえば、エージェントがプロジェクトを指定せずに課題を検索しようとした場合に、対象プロジェクトを自動的に絞り込むようなことができます。

class ScopeEnforcer(Middleware):
    async def on_call_tool(self, context: MiddlewareContext, call_next):
        args = context.message.arguments or {}
        # projectIdが未指定の場合、対象プロジェクトを自動的に絞り込む
        if context.message.name in ("get_issues", "count_issues", "search_issues"):
            if "projectId" not in args:
                context.message.arguments["projectId"] = [ALLOWED_PROJECT_ID]
        return await call_next(context)

前述の on_call_tool によるポリシー制御がリクエストの「拒否」であるのに対し、これはリクエストの「補正」です。 漫然と幅広な対象を処理してしまうことは防ぎつつ、クライアント側の明示指定による特定対象の処理は妨げない・・・といった具合のバランスをとることが可能です。

監査ログの記録

すべてのツール呼び出しをログに記録することで、Coding Agentが何を実行したかの監査証跡を残せます*6

import logging

logger = logging.getLogger("mcp_audit")

class AuditMiddleware(Middleware):
    async def on_call_tool(self, context: MiddlewareContext, call_next):
        logger.info(f"Tool called: {context.message.name}, args: {context.message.arguments}")
        result = await call_next(context)
        logger.info(f"Tool result: {context.message.name} completed")
        return result

なお、FastMCPにはビルトインの LoggingMiddleware も用意されています。

from fastmcp.server.middleware.logging import LoggingMiddleware

proxy.add_middleware(LoggingMiddleware(include_payloads=True))

複数MCPサーバーの統合

FastMCPの mount 機能を使えば、複数のMCPサーバーを1つのProxyサーバーに束ねることができます。

mcp = FastMCP(name="Unified Server")

backlog_proxy = create_proxy(backlog_config, name="Backlog")
github_proxy = create_proxy(github_config, name="GitHub")

mcp.mount(backlog_proxy, namespace="backlog")
mcp.mount(github_proxy, namespace="github")

クライアントからは1つのMCPサーバーに接続するだけで、backlog_get_issuegithub_list_repos といった複数サービスのツールを統一的に利用できます。

名前空間の付与 = ツールに対するサフィックスの付与により、同名異義のツール名の衝突を回避することも可能です

全体のコード

ここまでの内容を踏まえて、Backlog MCP Serverに対するProxy実装の全体像を示します。

from fastmcp import FastMCP
from fastmcp.server import create_proxy
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.exceptions import ToolError

mcp = FastMCP(name="Local Server")

config = {
    "mcpServers": {
        "backlog": {
            "command": "npx",
            "args": ["backlog-mcp-server"],
            "env": {
                "BACKLOG_DOMAIN": "yourspace.backlog.com",
                "BACKLOG_API_KEY": "your-api-key-here", # 実際のワークロードでは環境変数などを通じてセキュアに読み込むとよいです
                "ENABLE_TOOLSETS": "space,project,issue,wiki",
            },
        }
    }
}
backlog_proxy = create_proxy(config, name="Backlog")

HIGH_RISK_TOOL_PATTERNS = ("update", "add", "delete")
SANDBOX_PROJECT_IDS = (1111111111, 2222222222) # 実在しないProjectIdです


class BacklogPolicyMiddleware(Middleware):
    async def on_call_tool(self, context: MiddlewareContext, call_next):
        tool_name = context.message.name
        args = context.message.arguments or {}

        if tool_name.startswith(HIGH_RISK_TOOL_PATTERNS):
            if "projectId" not in args or args["projectId"] not in SANDBOX_PROJECT_IDS:
                raise ToolError(
                    f"Tool '{tool_name}' can only be called for sandbox projects."
                )

        return await call_next(context)

    async def on_list_tools(self, context: MiddlewareContext, call_next):
        tools = await call_next(context)
        HIDDEN_PREFIXES = (
            "add_project",
            "update_project",
            "delete_project",
            "delete_issue",
            "watching_list_items",
            "watching_list_count",
            "watching",
            "custom_fields",
            "milestone",
            "milestone_list",
            "delete_version",
        )
        filtered = [t for t in tools if not t.name.lower().endswith(HIDDEN_PREFIXES)]
        return filtered


backlog_proxy.add_middleware(BacklogPolicyMiddleware())
mcp.mount(backlog_proxy, namespace="backlog")


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

Coding Agentへの追加要領

冒頭でコメントした通り、上述のようなコードのProxyサーバーは自らのローカル環境に気軽に組み込んで利用することができます。

今回はClaude Codeで例を示しますが、その他のCoding Agentにおいても問題なく類似の組み込み方が可能です。

作成したProxyサーバーをClaude Codeに登録するには、claude mcp add コマンドを使います。

基本的にはこれだけで、先ほど説明したようなProxy機能を備えたMCPサーバーを使い始めることができます*7

基本的な追加方法

# uvxを使う場合(インストール不要、隔離環境で実行)
claude mcp add backlog -- uvx fastmcp run server.py

# 自環境でインストール済みのfastmcpを直接使う場合
claude mcp add backlog -- fastmcp run server.py

スコープの指定

--scope オプションで、設定の保存先を制御できます。

# プロジェクト共有(.mcp.json に保存、gitでコミット可能)
claude mcp add --scope project backlog -- uvx fastmcp run server.py

例えばチームで共有したいようなProxyの場合は--scope project を使い、.mcp.json 定義をProxyサーバーのコードともにリポジトリにコミットしておくようにすると便利でしょう。

登録の確認と削除

# 登録済みサーバーの一覧
claude mcp list

# サーバーの削除
claude mcp remove backlog

まとめ

既存のMCPサーバーが提供する権限制御の粒度が要件に合わない場合、FastMCPのProxy機能とMiddlewareを組み合わせることで、サーバーのコードを一切変更せずに入出力を制御できます。

  • on_list_tools でクライアントに公開するツールを絞り込む
  • on_call_tool で引数の内容に基づいてリクエストを許可・拒否する
  • create_proxymount により、複数のMCPサーバーを1つに統合することも可能

MCPサーバーを「そのまま使う」か「自前で書き直す」かの二択ではなく、Proxyで薄くラップする という第三の選択肢を持っておくと、運用の柔軟性、ツール活用の利便性/セキュリティの向上に寄与するケースはそれなりにあるでしょう。

脚注

*1:MCPサーバーによっては環境変数・リクエストヘッダーの内容等次第でツールセットを絞り込めるものもあります。ただし、その粒度はサーバーの実装次第であり、大まかなカテゴリ単位では指定できてもツール単位での細かい絞り込みには対応していないものは多いです

*2:プログラムの実装経験が豊富な方はDjangoやNext.jsあるいはLaravelといった著名Webフレームワークの同名機能を思い浮かべるかもしれませんが、凡そはそのイメージで問題ありません。ご存知ない方も既存のフレームワークのユースケースを調べることでこちらのユースケースのイメージを膨らませやすくなると思います

*3:上記の例はツール名のサフィックスで除外する拒否リスト方式ですが、逆に許可リスト方式で「使わせたいツールだけを明示的に列挙する」ことももちろんできます。むしろセキュリティの観点からは、許可リスト方式のほうがより安全と言えるかもしれません。MCPサーバーのアップデートで新しいツールが追加された場合、拒否リスト方式では意図せず公開されてしまいますが、許可リスト方式であれば明示的に追加しない限り公開されません

*4:実際はハードコーディングではなくローカルのyaml等を読ませて用いると便利でしょう。また、こうしたProxy機能を持つサーバーをリモートサーバーとしてホストする場合はJWT認証のようなものを前提として、個人の属性に応じて処理を制御することも有用と考えます

*5:他にはResourceErrorやPromptErrorというものもあります

*6:ローカルのロギングではなく、クラウド上の集約基盤に利用履歴を記録する形でももちろんいいでしょう

*7:ただし、各自の設定・工夫に頼らずに、確実に特定の制御を適用したい場合がエンタープライズのワークロードにはあると思います。その場合は、今回説明したようなFastMCPサーバーを適当な認証認可機構付きでクラウド上にデプロイしていただいたり、Amazon Bedrock AgentCore GatewayinterceptorsPolicyといった機能を組み入れつつ構築していただいたりしたのちに、remoteのMCPサーバーとして参照させるようにすればOKです

田斉 省吾 (記事一覧)

アプリケーションサービス部

2016年新卒入社。筆無精で入社以降一切ブログを書いてこなかったのですが、色々ありごく稀に書くことにしました(群馬から真心をこめてお届けします)