Safie Engineers' Blog!

Safieのエンジニアが書くブログです

カスタムMCPツールの制御について深堀る

メリー・クリスマス、セーフィーCTOの森本です。
この記事はSafie Engineers' Blog! Advent Calendar 25日目の記事です。

一昨年創業以来10年の開発組織の振り返りについて掲載し、2年が経ちます。
ありがたい事に会社の事業、開発組織も順調に新しい取り組みを交えつつ大きく成長している事を実感している今日この頃です。
これもメンバーの皆さんの頑張りの賜物であると感謝しています。
そのおかげもあって、自分の時間を手を動かす事に使うことも少しずつ出来るようになってきましたので、今回は少し技術よりの内容について書く事にします。

はじめに

生成AI周りでは様々なモデルやそれにまつわるツールの急速な発展により、業務効率化、創造性の向上、情報アクセスの簡易化などの点に於いて、驚くべきスピードで多種多様な恩恵がもたらされているのは今更言うまでもありません。
もちろん当社でも早い段階から、一部メンバーが様々なツールの開発業務への活用やプロダクトへの応用検討を進めてくれていましたが、これを更に加速する為チームを組成し、組織としての活用を進めています。

また、これも言うまでもないですが、MCP(Model Context Protocol)により多種多様な外部サービスと柔軟に連携ができるようになり、ACP(Agent Communication Protocol)、A2A(Agent-to-Agent)などによりエージェント間のコミュニケーションも広がりを見せています。
一例として当社社員が以下で紹介してくれていますが、柔軟なテキストベースのインプットによりサービスの制御が可能となっています。

MCPを活用すれば、LLMを経由して当社サービスを柔軟に制御する事が出来るだけでなく、他のサービスとの連携も非常に簡単に行えます。
上記記事にある通り、当社カメラから画像を取得しその要約をすることも簡単に出来ますし、要約した内容をインカムで通知すると言った他のサービスとの連携も簡単です。

例)インカムーSafie連携

今更ですが、一気にサービスの拡張性を伸ばしてくれる技術という事で強く可能性を感じています。

ただ、カスタムのMCPサーバーを活用する為にはClaude Desktopが必要なの(他にも候補はありますが)がちょっと勿体ないのと、内部的にLLMとMCPサーバーがどのようにやり取りしているのか知りたかったので、MCPサーバーを自前のサーバー上で制御できる簡易なサンプルを作り中身を見てみました。

MCPサーバー活用例 w/ Claude Desktop

Claude Desktopでは入力されたプロンプトによって、柔軟にMCPサーバーが提供するツールが使えるかどうか判断し、必要に応じ適切にパラメータをセットし実行してくれます。

例えば以下のようなカスタムのMCPサーバーを準備して、Claude Desktopに組み込み(※このあたりの手順は省きます)、「こんばんはとあいさつしてね」とClaude Desktopのプロンプトを入力すると、MCPサーバーの提供する機能を適切に実行し、そのレスポンスを返してくれます。

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field

class GreetingMessage(BaseModel):
   message: str = Field(None, description="あいさつのメッセージ")

# Create an MCP server
mcp = FastMCP("Greeting")

@mcp.tool()
def greeting(message: str | None = Field(
       description="あいさつのメッセージ", default=None
   )) -> GreetingMessage:
   """
   あいさつメッセージを返します
   """
   return GreetingMessage(message=f"Hello, MCP! {message}")

def run():
   mcp.run()

よく知られた挙動ですが、内部でどうやっているか見てみました。

LLMとMCPサーバーのやり取り

MCPサーバーを使用しない、従来のやり取り

※自分なりの解釈が相当入っていますが、そこはご容赦ください。
※llmプロキシとしてLiteLLMを利用し、モデルはclaude-sonnet-4
を使用しています。
最初に従来のLLMへのチャットメッセージは以下のようなリクエストとレスポンスで成立しています。

response = client.chat.completions.create(
    model=request.model,
    messages=messages,
    tools=mcp_manager.tools if request.use_mcp else None,
    max_tokens=request.max_tokens,
    temperature=request.temperature,
)
messages = [
  {
    "role": "assistant",
    "content": "あなたは柔軟に入力内容を類推できるAIアシスタントです。質問への回答は出来る限り完結に回答してくれます。"
  },
  {
    "role": "user",
    "content": "こんばんはとあいさつしてね"
  }
]
tools = []

※LLMに出来る限り簡潔にレスポンスを返して欲しかったので、上記のようなアシスタントロールのメッセージを最初にセットしています。

すると以下のようなレスポンスを返してくれます。

ChatCompletion(
    id='chatcmpl-b49350d7-d24a-43c4-a2a3-77343982f33d',
    choices=[
        Choice(
            finish_reason='stop',
            index=0,
            logprobs=None,
            message=ChatCompletionMessage(
                content='こんばんは!',
                refusal=None,
                role='assistant',
                annotations=None,
                audio=None,
                function_call=None,
                tool_calls=None
            )
        )
    ],
    created=1766406680,
    model='apac.anthropic.claude-sonnet-4-20250514-v1:0',
    object='chat.completion',
    service_tier=None,
    system_fingerprint=None,
    usage=CompletionUsage(
        completion_tokens=9,
        prompt_tokens=83,
        total_tokens=92,
        completion_tokens_details=None,
        prompt_tokens_details=PromptTokensDetails(
            audio_tokens=None,
            cached_tokens=0,
            cache_creation_tokens=0
        ),
        cache_creation_input_tokens=0,
        cache_read_input_tokens=0
    )
)

LLMがchatAPIで渡されたプロンプトメッセージ「こんばんはとあいさつしてね」に対して、「こんばんは!」というメッセージを返しています。

MCPサーバーを介する場合のやり取り

上記に対し、LLMが入力されたプロンプトを元に最適なMCPツールを選択し必要に応じてパラメータをセットし実行するためには、サポートしているMCPツールの情報をインプットしてやる必要があります。

messages =   {
    "role": "assistant",
    "content": "あなたは柔軟に入力内容を類推できるAIアシスタントです。質問への回答は出来る限り完結に回答してくれます。"
  },
tools = [
  {
    "type": "function",
    "function": {
      "name": "greeting",
      "description": "あいさつメッセージを返します",
      "parameters": {
        "type": "object",
        "properties": {
          "message": {
            "type": "string",
            "description": "message parameter",
            "default": null
          }
        },
        "required": []
      }
    }
  }
]

先程との違いは明確で、toolsパラメーターにMCPサーバーが提供するツールの情報をセットして、chatAPIを実行しています。

すると以下のようなレスポンスを返してくれます。

ChatCompletion(
    id='chatcmpl-63876e3e-fb1d-4910-81a9-8fea53190dff',
    choices=[
        Choice(
            finish_reason='tool_calls',
            index=0,
            logprobs=None,
            message=ChatCompletionMessage(
                content='',
                refusal=None,
                role='assistant',
                annotations=None,
                audio=None,
                function_call=None,
                tool_calls=[
                    ChatCompletionMessageFunctionToolCall(
                        id='tooluse_rSY9i9tdRKmTZewG_-lfqg',
                        function=Function(
                            arguments='{"message": "こんばんは"}',
                            name='greeting'
                        ),
                        type='function',
                        index=0
                    )
                ]
            )
        )
    ],
    XXXXXXXXXXXXXXXX
)

レスポンスも先程とは異なり、tool_callsパラメータがセットされています。
パラメータにはchatAPIで渡したプロンプトとtoolsパラメータの内容を元にLLMが選択した最適な実行すべきMCPツールが設定されています。
LLMはあくまで実行すべきツールを提示してくれるだけなので、実際のツール実行部分はLLMの範疇外なので対応が必要です。
上記ではgreetingというツールをmessageに記載されているパラメータで実行すべきとなっているので、実行します。

{
 'message': 'Hello, MCP! こんばんは'
}

実行結果は自身で準備したMCPツールの通りとなります。

今回はレスポンスとして「Hello, MCP! こんばんは」が返ってきていますが、
プロンプトから入力したテキストが正しくパラメータとして渡され、ツールも想定通りに実行されています。
※あくまでLLMは実行すべきと判断した内容を返してくれるだけで、ツールの実行は自身でやっています。

次にLLMにMCPツールの実行結果をチャットAPIで渡してやります。

messages = [
  {
    "role": "assistant",
    "content": "あなたは柔軟に入力内容を類推できるAIアシスタントです。質問への回答は出来る限り完結に回答してくれます。"
  },
  {
    "role": "user",
    "content": "こんばんはとあいさつしてね"
  },
  {
    "role": "assistant",
    "content": "",
    "tool_calls": [
      {
        "id": "tooluse_rSY9i9tdRKmTZewG_-lfqg",
        "type": "function",
        "function": {
          "name": "greeting",
          "arguments": "{\"message\": \"こんばんは\"}"
        }
      }
    ]
  },
  {
    "role": "tool",
    "tool_call_id": "tooluse_rSY9i9tdRKmTZewG_-lfqg",
    "content": "{\"message\": \"Hello, MCP! こんばんは\"}"
  }
]

※toolsは重複するので記載を省きます。

LLMが前のステップで返してきたtool_call_idに合わせてツールのレスポンスをセットしチャットメッセージを実行します。これによりLLMがMCPツールが実行された事と、その結果を受け取る事ができます。

ChatCompletion(
    id='chatcmpl-32ae0203-9b39-4b33-9a71-6ba9c9ac1cdb',
    choices=[
        Choice(
            finish_reason='stop',
            index=0,
            logprobs=None,
            message=ChatCompletionMessage(
                content='こんばんは!いかがお過ごしですか?',
                refusal=None,
                role='assistant',
                annotations=None,
                audio=None,
                function_call=None,
                tool_calls=None
            )
        )
    ],
    XXXXXXXXXXXXXXXX
)

するとLLMが今までのやり取りとMCPツールの実行結果をベースにレスポンスを返してくれます。
今回は「こんばんは!いかがお過ごしですか?」です。若干例が分かりにくいですね。笑。

更に連続でMCPツールが実行出来る場合はtool_callsに次に実行すべきMCPツールの情報がセットされるのですが、今回は特に設定されて無いのでこれでチャットプロセスは完了となります。

MCPツールをカスケードで活用する場合

では冒頭でご紹介した、当社メンバーが作成したSafie API MCPツールを使って、カスケードでMCPツールを実行する例を見てみましょう。

先程のMCPツールも読み込みつつ、「昨日の10時の画像に人が映っているかどうか調べて」というプロンプトを入力します。

尚、時間情報が含まれる場合はタイムゾーンを解釈するようなプロンプトメッセージを付加しています。
いきなり長くなりますが、LLMへ投げるプロンプトは以下の通りとなります。

messages = [
  {
    "role": "assistant",
    "content": "あなたは柔軟に入力内容を類推できるAIアシスタントです。質問への回答は出来る限り完結に回答してくれます。\n\n## 日時情報\n現在の日時: 2025-12-22T13:51:54.528550+00:00\n現在の日付: 2025-12-22\n\n相対的な日付表現(今
日、明日、昨日など)は上記の現在日時を基準として\n言語から類推できるタイムゾーンに変換の上\nISO 8601拡張形式(タイムゾーン付き)で解釈してください。"
  },
  {
    "role": "user",
    "content": "昨日の10時の画像に人が映っているかどうか調べて"
  }
]
tools = [
    {
      "type": "function",
      "function": {
        "name": "list_devices",
        "description": "    アクセス権限のあるデバイスの一覧を取得します    ",
        "parameters": {
          "type": "object",
          "properties": {
            "item_id": {
              "type": "integer",
              "description": "item_id parameter",
              "default": null
            }
          },
          "required": []
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "get_device_image",
        "description": "    指定されたデバイスから画像を取得します    timestampを指定しない場合、API実行時点の最新画像が取得できます    ",
        "parameters": {
          "type": "object",
          "properties": {
            "device_id": {
              "type": "string",
              "description": "device_id parameter"
            },
            "timestamp": {
              "type": "string",
              "description": "timestamp parameter",
              "default": null
            }
          },
          "required": ["device_id"]
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "list_device_media",
        "description": "    指定されたデバイスで録画されている映像(メディア)の一覧を取得します    timestampを指定しない場合、API実行時点の最新画像が取得できます    制限:    - start/endの取得最大範囲は 1日(86400秒)です    ",
        "parameters": {
          "type": "object",
          "properties": {
            "device_id": {
              "type": "string",
              "description": "device_id parameter"
            },
            "start": {
              "type": "string",
              "description": "start parameter"
            },
            "end": {
              "type": "string",
              "description": "end parameter"
            }
          },
          "required": ["device_id", "start", "end"]
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "get_device_location",
        "description": "    指定されたデバイスの現在のGPS位置情報を取得します。GPSに対応していないデバイスでは利用できません    デバイスに手動設定された位置情報を取得することはできません    ",
        "parameters": {
          "type": "object",
          "properties": {
            "device_id": {
              "type": "string",
              "description": "device_id parameter"
            }
          },
          "required": ["device_id"]
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "get_device_thumbnail",
        "description": "    指定されたデバイスの最新サムネイルを取得します    ",
        "parameters": {
          "type": "object",
          "properties": {
            "device_id": {
              "type": "string",
              "description": "device_id parameter"
            }
          },
          "required": ["device_id"]
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "list_device_standard_events",
        "description": "    指定されたデバイスの標準イベント情報一覧を取得します    「標準イベント」とは以下の5つのイベントの総称です    - 接続検知    - 切断検知    - モーション検知    - サウンド検知    - 人検知    制限:    - start/endの取得最大範囲は 1日(86400秒)です    ",
        "parameters": {
          "type": "object",
          "properties": {
            "device_id": {
              "type": "string",
              "description": "device_id parameter"
            },
            "start": {
              "type": "string",
              "description": "start parameter"
            },
            "end": {
              "type": "string",
              "description": "end parameter"
            },
            "event_types": {
              "type": "array",
              "enum": ["connect", "disconnect", "motion", "sound", "person"],
              "description": "event_types parameter connect, disconnect, motion, sound, person",
              "default": null
            }
          },
          "required": ["device_id", "start", "end"]
        }
      }
    },
    {
      "type": "function",
      "function": {
        "name": "greeting",
        "description": "    あいさつメッセージを返します    ",
        "parameters": {
          "type": "object",
          "properties": {
            "message": {
              "type": "string",
              "description": "message parameter",
              "default": null
            }
          },
          "required": []
        }
      }
    }
]

入力が時間情報を含むプロンプトの場合は、タイムゾーンやフォーマットを指定するメッセージを補完しています。
toolsパラメータが長くなっていますが、こちらはサポートするMCPツールをリスト化して設定しています。
こちらのメッセージに対するレスポンスは以下となります。

ChatCompletion(
    id='chatcmpl-211b219f-f4af-4e42-8ac1-c6195b2ed713',
    choices=[
        Choice(
            finish_reason='tool_calls',
            index=0,
            logprobs=None,
            message=ChatCompletionMessage(
                content='昨日の10時の画像を調べるために、まずアクセス可能なデバイスの一覧を取得します。',
                refusal=None,
                role='assistant',
                annotations=None,
                audio=None,
                function_call=None,
                tool_calls=[
                    ChatCompletionMessageFunctionToolCall(
                        id='tooluse_QDchcs-OTn60bSHhwYflUw',
                        function=Function(
                            arguments='{}',
                            name='list_devices'
                        ),
                        type='function',
                        index=1
                    )
                ]
            )
        )
    ],
    XXXXXXXXXXXXXXXX
)

LLMは指定されたプロンプトを実行するために、まずはアクセス可能なカメラの一覧取得を行うべきと判断し、list_deviceAPIを実行するようレスポンスを返してくれます。素晴らしく柔軟ですね!!

先程の例と同じで、実行すべきMCPツールのメソッドが指定されているので、そのとおりに実行します。

{
    'result': [
        {
            'device_id': 'xxxxxxxx',
            'serial': 'xxxxxxxx',
            'setting': {
                'name': 'XXXXXXXX'
            },
            'status': {
                'video_streaming': True
            },
            'model': {
                'description': 'One (SF-1)'
            }
        }
    ]
}

指定された通りにMCPツールのメソッドを実行し上記のレスポンスを得ます。
同様にこちらのレスポンスを含めてLLMにチャットを送信します。

messages = [
    {
        'role': 'assistant',
        'content': 'あなたは柔軟に入力内容を類推できるAIアシスタントです。質問への回答は出来る限り完結に回答してくれます。\n\n## 日時情報\n現在の日時: 2025-12-22T14:05:27.941094+00:00\n現在の日付: 2025-12-22\n\n相対的な日付表現(今日、明日、昨日など)は上記の現在日時を基準として\n言語から類推できるタイムゾーンに変換の上\nISO 8601拡張形式(タイムゾーン付き)で解釈してください。'
    },
    {
        'role': 'user',
        'content': '昨日の10時の画像に人が映っているかどうか調べて'
    },
    {
        'role': 'assistant',
        'content': '昨日の10時の画像を調べるために、まずアクセス可能なデバイスの一覧を取得させていただきます。',
        'tool_calls': [
            {
                'id': 'tooluse_zVu7PAOFSHqr5GYrKxGM7A',
                'type': 'function',
                'function': {
                    'name': 'list_devices',
                    'arguments': '{"item_id": null}'
                }
            }
        ]
    },
    {
        'role': 'tool',
        'tool_call_id': 'tooluse_zVu7PAOFSHqr5GYrKxGM7A',
        'content': '{"result": [{"device_id": "xxxxxxxx", "serial": "xxxxxxxx", "setting": {"name": "XXXXXXXX"}, "status": {"video_streaming": true}, "model": {"description": "One (SF-1)"}}]}'
    }
]

※toolsは重複するので記載を省きます。

上記メッセージへのレスポンスは以下となります。

ChatCompletion(
    id='chatcmpl-be9de40c-a119-4548-930a-b75abff8f898',
    choices=[
        Choice(
            finish_reason='tool_calls',
            index=0,
            logprobs=None,
            message=ChatCompletionMessage(
                content='デバイスが1台見つかりました。昨日(2025-12-21)の10時の画像を取得して確認します。',
                refusal=None,
                role='assistant',
                annotations=None,
                audio=None,
                function_call=None,
                tool_calls=[
                    ChatCompletionMessageFunctionToolCall(
                        id='tooluse_dGHdMM4PQt-fAdNBmMTEhg',
                        function=Function(
                            arguments='{"device_id": "xxxxxxxx", "timestamp": "2025-12-21T10:00:00+09:00"}',
                            name='get_device_image'
                        ),
                        type='function',
                        index=1
                    )
                ]
            )
        )
    ],
    XXXXXXXXXXXXXXXX
)

すると今度はtool_callsに次に実行すべきMCPツールが指定されています。
get_device_imageを指定のパラメータ付きで実行しろと言うことですね。

言われたとおりにツールを実行します。
get_device_imageは指定されたパラメータを元に、サムネイルデータを取得します。
取得したサムネイルデータをBase64形式に変換し、先程の例と同様にLLMにメッセージとして入力します。(※画像データは重いので記載しません)

messages = [
    {
        'role': 'assistant',
        'content': 'あなたは柔軟に入力内容を類推できるAIアシスタントです。質問への回答は出来る限り完結に回答してくれます。\n\n## 日時情報\n現在の日時: 2025-12-22T14:05:27.941094+00:00\n現在の日付: 2025-12-22\n\n相対的な日付表現(今日、明日、昨日など)は上記の現在日時を基準として\n言語から類推できるタイムゾーンに変換の上\nISO 8601拡張形式(タイムゾーン付き)で解釈してください。'
    },
    {
        'role': 'user',
        'content': '昨日の10時の画像に人が映っているかどうか調べて'
    },
    {
        'role': 'assistant',
        'content': '昨日の10時の画像を調べるために、まずアクセス可能なデバイスの一覧を取得させていただきます。',
        'tool_calls': [
            {
                'id': 'tooluse_zVu7PAOFSHqr5GYrKxGM7A',
                'type': 'function',
                'function': {
                    'name': 'list_devices',
                    'arguments': '{"item_id": null}'
                }
            }
        ]
    },
    {
        'role': 'tool',
        'tool_call_id': 'tooluse_zVu7PAOFSHqr5GYrKxGM7A',
        'content': '{"result": [{"device_id": "xxxxxxxx", "serial": "xxxxxxxx", "setting": {"name": "XXXXXXXX"}, "status": {"video_streaming": true}, "model": {"description": "One (SF-1)"}}]}'
    },
    {
        'role': 'assistant',
        'content': 'デバイスが1台見つかりました。昨日(2025-12-21)の10時の画像を取得して確認します。',
        'tool_calls': [
            {
                'id': 'tooluse_dGHdMM4PQt-fAdNBmMTEhg',
                'type': 'function',
                'function': {
                    'name': 'get_device_image',
                    'arguments': '{"device_id": "xxxxxxxx", "timestamp": "2025-12-21T10:00:00+09:00"}'
                }
            }
        ]
    },
    {
        'role': 'tool',
        'tool_call_id': 'tooluse_dGHdMM4PQt-fAdNBmMTEhg',
        'content': 'get_device_image'
    },
    {
        'role': 'user',
        'content': [
            {
                'type': 'image_url',
                'image_url': {
                    'url': 'data:image/jpeg;base64,XXXXXXXX'
                }
            },
            {
                'type': 'text',
                'text': '以降のMCPツール呼び出しで画像が必要な場合は、画像データをMCPツールのargumentsに設定してください。'
            }
        ]
    }
]

※toolsは重複するので記載を省きます

レスポンスは以下の通りです。

ChatCompletion(
    id='chatcmpl-57a6d457-306a-4f27-b14e-a4696f11172d',
    choices=[
        Choice(
            finish_reason='stop',
            index=0,
            logprobs=None,
            message=ChatCompletionMessage(
                content='昨日(2025-12-21)の10時頃の画像を確認しました。\n\n画像を見る限り、**人は映っていません**。\n\n画像には以下が写っています:\n- 赤い自動販売機\n- 商品が陳列された木製の棚\n- 白いタイル張りの床\n- 右下に白い機器(エアコンの室外機のようなもの)\n\n店舗や事務所のような場所の監視カメラの映像のようですが、この時点では人の姿は確認できませんでした。',
                refusal=None,
                role='assistant',
                annotations=None,
                audio=None,
                function_call=None,
                tool_calls=None
            )
        )
    ],
    XXXXXXXXXXXXXXXX
)

これ以上MCPツールを実行する必要が無いので、tool_callsはNoneとなっています。
また、LLMは設定した画像データをベースに適切にプロンプトで指定された内容を処理したレスポンスを返しています。

入力されたプロンプトに基づいて、MCPサーバーを設定しつつ最適なレスポンスを導き出す手法は上記の繰り返しです。

こうやってみると適切な情報をセットしてやる必要はありますが、それらの情報を元にLLMが如何に柔軟にテキスト情報を処理し、必要に応じ最適なMCPツールを選択しレスポンスを生成しているかがよく分かります。
同様に様々なサービスのMCPツールを組み込むだけで、容易にLLMを介したサービス間自動連携が実現できます。

MCPツールの情報をLLMに設定するには

各メソッドの情報を抽出し、登録する必要があります。
参考程度ですが、軽く書いてみたので貼っておきます。
複数のMCPサーバーを統合管理できるようにもしてみました。

import json
import logging
from mcp.server.fastmcp import FastMCP

class MCP_Manager():
   def __init__(self):
       #self._mcps = {}
       self._tools = {}
       self.logger = logging.getLogger(__name__)

   async def register_tool(self, mcp: FastMCP):
       """指定されたMCPサーバーをMcpManagerに登録する"""
       for tool in await mcp.list_tools():
           doc = tool.description
           description = doc.replace('\n', '') if doc else f"{tool.name}"
           properties = {}

           # 簡単なパラメータ推論(実際にはより詳細な実装が必要)
           required = tool.inputSchema.get("required", [])
           for param_name, param in tool.inputSchema["properties"].items():
               self.logger.info(f"Param : {param}")
               if param_name != 'self':
                   param_type = "string" # デフォルト
                   # 一旦地道に実装する(もっとよいやり方があると思うが)
                   _any = param.get("anyOf")
                   if _any is not None:
                       param = _any[0]
                       if "items" in param.keys():
                           def_name = param["items"]["$ref"].split("/")[-1]
                           properties[param_name] = {
                               "type": param["type"],
                               "enum": tool.inputSchema["$defs"][def_name]["enum"],
                               "description": f"{param_name} parameter {', '.join(tool.inputSchema["$defs"][def_name]["enum"])}"
                           }
                       else:
                           properties[param_name] = {
                               "type": param["type"],
                               "description" : param["description"] if description in param.keys() else f"{param_name} parameter",
                           }
                   else:
                       if param["type"] == int:
                           param_type = "integer"
                       elif param["type"] == float:
                           param_type = "number"
                       elif param["type"] == bool:
                           param_type = "boolean"

                       properties[param_name] = {
                           "type": param_type,
                           "description" : param["description"] if description in param.keys() else f"{param_name} parameter",
                       }
                   if param_name not in required:
                       # Default値は現状は参考程度にしかならないが一応セット
                       properties[param_name]["default"] = param.get("default", None)

           self._tools[tool.name] = {
               "mcp": mcp,
               "name": tool.name,
               "description": description,
               "properties": properties,
               "required": required
           }
           self.logger.info(f"Registered tool: {tool.name} with properties: {properties} and required: {required}")
          
             @property
   def tools(self):
       """登録されたツールの一覧をBedrock API仕様に準拠する形で返却する"""
       return [{"type": "function",
               "function": {
                   "name": v["name"],
                   "description": v["description"],
                   "parameters": {
                       "type": "object",
                       "properties": v["properties"],
                       "required": v["required"]
                   }
               }} for k,v in self._tools.items()]
  
   async def call_tool(
           self,
           tool_name: str,
           tool_args: dict):
       """指定されたツールを実行し、結果を返す"""
       self.logger.info(f"Function call detected: {tool_name}")
       # FastMCPツールを実行
       if tool_name in self._tools.keys():
           try:
               # パラメーターの補完を行う (default値はclaudeでは補完されない)
               for k,v in self._tools[tool_name]["properties"].items():
                   if k not in tool_args.keys():
                       if "default" in v:
                           tool_args[k] = v["default"]
               self.logger.info(f"Executing tool: {tool_name} with args {tool_args}")

               # 後で必要性に応じ修正する
               tool_result = await self._tools[tool_name]["mcp"].call_tool(
                   tool_name,
                   tool_args
               )
               self.logger.info(f"Tool Result: {tool_result} Type: {type(tool_result)} Len: {len(tool_result)}")

               return tool_result

           except Exception as e:
               self.logger.error(f"Tool execution error: {e}")
               error_message = f"ツール '{tool_name}' の実行中にエラーが発生しました: {str(e)}"
               #return error_message
               raise Exception(error_message)
          
       return {}

まとめ

  • MCPの活用により、様々なツール、サービスを柔軟且つ統一的に活用する事が可能となります。
  • 更に最近ではより上位のレイヤーでのコミュニケーションが可能となるエージェント間コミュニケーション・プロトコルも流行ってきている。
    • 主要な候補としてACP、A2Aが存在していますが、今後A2Aに統合されるような話もあるようです。
  • これらの活用により、生成AIの活用の幅が更に広がる事が期待されます。
  • 尚、上記の実装例も生成AIを活用して作成しています。(本当に便利です)
    • 一方でそのままでは活用できず、細かな修正を加える必要はありました。
    • ※やはり盲目的に使うのではなく、内容を理解しつつ適切に活用すべきだと改めて感じています。

最期に

  • セーフィーでは開発効率向上やプロダクトの価値向上に向けて、積極的に生成AI活用の検討も進めています。
  • 最新の技術も組み合わせつつ、更なる成長へ向けて様々な開発に関わる取り組みを行っています。それらに一緒に関わってくれるエンジニアさんを絶賛募集しています!!!!

safie.co.jp

© Safie Inc.