はじめに
前回の投稿では、MCP Server & Client を FastMCP を使って実装し、MCP Server を Cline から動かしてみました。
しかしながら、お題が 「OCI のメトリクスを参照する」というものだったので、ちょっと物足りない... やっぱり、何か対象に対して働きかけるものでないと Agentっぽくないかな？ ということで、今回は OCI の Compute インスタンスの上げ下げを自然言語で実行できる MCP Server を作ってみたいと思います。
ただし、この手の更新系の MCP Server で怖いのは、LLM のハルシネーションや悪意あるプロンプトによって想定外の指示が MCP Server に渡り、その結果、悲惨な状況になってしまう恐れがあるということです。
MCP には Elicitation と呼ばれる、サーバーがユーザーとのインタラクション中にクライアントを通じてユーザーから追加情報を要求するための標準化された方法があります。
サーバーはクライアント経由でユーザーにメッセージと入力スキーマを提示し、ユーザーの入力に応じて「accept」（受け入れ）、「decline」（拒否）、「cancel」（キャンセル）というアクションと、任意で構造化された回答内容を受け取ることができるよう設計されています。
ということで、今回のお題は、「ユーザーに yes/no の確認を行なった後に Compute インスタンスの上げ下げを行う」MCP Server を作ってみたいと思います。今回は、VS Code のワークスペースに MCP Server を定義して GitHub Copilot Chat から呼び出してみます。
MCP Server を作成する
MCP のライブラリは FastMCP です。Elicitation に対応したバージョンを使って下さい。
import sys, os
import oci
from oci.core import ComputeClient
from oci.core.models import Instance
from fastmcp import FastMCP, Context
from typing import Annotated
from pydantic import BaseModel, Field
# Create an MCP server
mcp = FastMCP("ComputeControl")
# list() 対象のリージョン
regions = ["us-ashburn-1", "us-phoenix-1", "ap-tokyo-1", "ap-osaka-1"]
# list() 対象のコンパートメント
compartments = {
"ocid1.compartment.oc1..aaaaaaaa4occ..." : "foo",
"ocid1.compartment.oc1..aaaaaaaansrw..." : "boo"
}
config = oci.config.from_file()
expose_instance_id = os.getenv("EXPOSE_INSTANCE_ID", "false").lower() == "true"
enable_elicitation = os.getenv("ENABLE_ELICITATION", "false").lower() == "true"
# list() の返り値となるクラス
class ComputeInstance(BaseModel):
name: str = Field(description="name of instance")
id: str | None = Field(default=None, description="id of instance")
region: str = Field(description="region where instance is located")
state: str = Field(description="current state of instance")
# Elicitation で Client から返される入力値を保持するクラス
class UserInput(BaseModel):
value: str = Field(description="reconfirmation of the action to take")
@mcp.tool()
async def list() -> list[ComputeInstance]:
"""
Compute インスタンスを全て検索する、ComputeInstance のリストを返す
ComputeInstanceクラス は name, id, region, state を含む
"""
all_instances = []
for region in regions:
compute_client = ComputeClient(
config=oci.config.from_file(),
service_endpoint=f"https://iaas.{region}.oraclecloud.com"
)
for compartment_id in compartments.keys():
try:
instances = compute_client.list_instances(compartment_id).data # type: list[Instance]
for instance in instances:
instance_id = instance.id if expose_instance_id else None
all_instances.append(ComputeInstance(
name=instance.display_name,
id=instance_id,
region=region,
state=instance.lifecycle_state
))
except Exception as e:
pass
return all_instances
@mcp.tool()
async def action(
ctx: Context,
region: Annotated[str, Field(description="region of the instance")],
action_to_take: Annotated[str, Field(description="action to take, either 'start' or 'stop'")],
id: Annotated[str, Field(description="id of the instance")] | None = None,
name: Annotated[str, Field(description="display name of the instance")] | None = None,
) -> str:
"""
Compute instance を 起動・停止 する
id か name かのどちらかを指定する必要がある
idを指定する方が望ましい
nameを指定した場合、nameからidを特定するが、複数の候補があった場合は処理を中断する
id, name, region は 最初に list() を実行してインスタンスの正確な情報を取得するべきである
"""
compute_client = ComputeClient(
config=oci.config.from_file(),
service_endpoint=f"https://iaas.{region}.oraclecloud.com"
)
if id:
instance = compute_client.get_instance(id).data # type: Instance
else:
if not name:
return f"id か name か、どちらかのパラメータが必要です"
target_instances = [] # list[Instance]
for compartment_id in compartments.keys():
try:
instances = compute_client.list_instances(compartment_id).data # type: list[Instance]
for instance in instances:
if instance.display_name == name:
target_instances.append(instance)
except Exception as e:
pass
if len(target_instances) == 0:
return f"{name} というインスタンスが見つかりませんでした"
if len(target_instances) > 1:
# 本当はユーザーに選択させたいけど、長くなるので今回は割愛
return f"{name} というインスタンスが複数あるので処理を中止します"
else:
instance = target_instances[0]
id = instance.id
name = instance.display_name
state = instance.lifecycle_state
msg = None
if action_to_take.lower() == 'start' and state != "STOPPED":
msg = f"{name} は \"STOPPED\" のステータスでない (\"{state}\") ので起動はやめときます"
elif action_to_take.lower() == 'stop' and state != "RUNNING":
msg = f"{name} は \"RUNNING\" のステータスでない (\"{state}\") ので停止はやめときます"
else:
if enable_elicitation:
# ユーザーの入力を確認
act = "起動" if action_to_take == "start" else "停止"
result = await ctx.elicit(
message = f'{name} のステータスは {state} です. {act}しますか? [y/N]',
response_type = UserInput # Class を指定する
)
print(f"result: {result}", file=sys.stderr)
if result.action == "accept":
if result.data and result.data.value.lower() in ("y", "yes"):
pass
else:
msg = f"処理を中止しました (by user input: {result.data.value})"
elif result.action == "decline":
msg = "処理を中止しました (decline)"
else: # cancel
msg = "処理を中止しました (cancel)"
if not msg:
if action_to_take.lower() == 'start':
response = compute_client.instance_action(instance.id, "START").data # type: Instance
msg = f"{name} を起動しました、ステータスは \"{response.lifecycle_state}\" です"
elif action_to_take.lower() == 'stop':
response = compute_client.instance_action(instance.id, "SOFTSTOP").data # type: Instance
msg = f"{name} を停止しました、ステータスは \"{response.lifecycle_state}\" です"
else:
msg = f"無効なアクション: {action_to_take}"
return msg
if __name__ == "__main__":
mcp.run(transport="stdio")
list() と action() という二つの tool を定義しています。
-
list()
Compute インスタンスをリストします。name, id, region, state を返します。
環境変数 EXPOSE_INSTANCE_ID によって、返り値に id を含めるかどうかを制御することができます。デフォルトでは LLM にインスタンスID を渡さないようにしています。 -
action()
インスタンスの起動・停止を行います。
id を使えばインスタンスが一意に決まりますが、id の指定がなくても name で対象を絞り込みます。
環境変数 ENABLE_ELICITATION が true だった場合、Client に{name} のステータスは {state} です. {act}しますか? [y/N]というメッセージを送り、ユーザーの入力を Client から返してもらい、その結果によって後続の処理を分岐します。この環境変数はまだまだ Elicitation の対応ができていないツールが多い状況への配慮です。
Elicitaion の実装部分をもう少し解説しましょう。
result = await ctx.elicit(
message = f'{name} のステータスは {state} です. {act}しますか? [y/N]',
response_type = UserInput # Class を指定する
)
response_type はちょっとトリッキーなのですが、スカラ型か pydantic の BaseModel を継承した任意のクラスを指定すればOKです。
FastMCPのドキュメンテーション に、このあたりの仕組みが書かれています。MCP 仕様が定めるスキーマ規定に従ったJSON オブジェクトで Server - Client 間の通信を行わないといけないのですが、FastMCP ではこのあたりをうまくラップする仕組みを提供しているので、スカラ型(str や bool など)を指定しても問題なく動作します。
今回は、UserInput というクラスを作成して、これを使ってユーザー入力を受け渡ししています。UserInput クラスには value というフィールドしかありませんが、クラスに入力項目を複数定義しても構いません。
class UserInput(BaseModel):
value: str = Field(description="reconfirmation of the action to take")
MCP Client を作成する
MCP Server をテストするための MCP Client を作ります。
import asyncio, argparse, sys
from fastmcp import Client
from fastmcp.client.transports import PythonStdioTransport
from fastmcp.client.client import CallToolResult
from fastmcp.client.elicitation import ElicitResult
# Elicitation のコールバックを受けるハンドラ
async def elicitation_handler(message: str, response_type: type, params, context):
print(f"response_type: {response_type}", file=sys.stderr)
user_input = input(f"{message}: ") # ユーザからの入力
response_data = response_type(value=user_input) # UserInput型のオブジェクトを作成
return response_data # return ElicitResult(action="accept", content=response_data) と同値
async def compute_control(args):
# stdio - MCP Server を子プロセスとして起動
server_script = "compute-control.py"
env = {} # 環境変数を子プロセスに渡す
if args.enable_elicitation:
env["ENABLE_ELICITATION"] = "true"
if args.expose_instance_id:
env["EXPOSE_INSTANCE_ID"] = "true"
transport = PythonStdioTransport(script_path=server_script, env=env)
# クライアントを作成してサーバと通信、Elicitation のハンドラを指定する
async with Client(transport, elicitation_handler=elicitation_handler) as client:
# インスタンスの一覧を取得する
#result = await client.call_tool("list") # type: CallToolResult
#print(result)
# インスタンスを起動・停止する
result = await client.call_tool("action",
{
"id": args.id,
"name" : args.name,
"region": args.region,
"action_to_take": args.action
}) # type: CallToolResult
print(result.data)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--region", required=True)
parser.add_argument("--action", required=True)
parser.add_argument("--enable-elicitation", action='store_true')
parser.add_argument("--expose-instance-id", action='store_true')
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--id")
group.add_argument("--name")
args = parser.parse_args()
asyncio.run(compute_control(args))
if __name__ == "__main__":
main()
Client を作成する際に、Server からの Elicitation コールバックを処理するハンドラを指定しています。
async with Client(transport, elicitation_handler=elicitation_handler) as client:
ハンドラ elicitation_handler 関数ですが、ここの返り値は本来、以下のように ElicitResult を返すべきですが、上の実装のように response_type 型のオブジェクトをそのまま返しても問題ありません、その場合は action = "accept" とみなされて、返り値が自動的に ElicitResult にラップされます。
response_data = response_type(value=user_input) # UserInput型のオブジェクトを作成
return ElicitResult(action="accept", content=response_data)
では、実行してみましょう。
Ashburn にある panda という名前のインスタンスを起動します。
$ python compute-control-client.py \
--name panda --region us-ashburn-1 --action start --enable-elicitation
╭─ FastMCP 2.0 ──────────────────────────────────────────────────────────────╮
│ │
│ _ __ ___ ______ __ __ _____________ ____ ____ │
│ _ __ ___ / ____/___ ______/ /_/ |/ / ____/ __ \ |___ \ / __ \ │
│ _ __ ___ / /_ / __ `/ ___/ __/ /|_/ / / / /_/ / ___/ / / / / / │
│ _ __ ___ / __/ / /_/ (__ ) /_/ / / / /___/ ____/ / __/_/ /_/ / │
│ _ __ ___ /_/ \__,_/____/\__/_/ /_/\____/_/ /_____(_)____/ │
│ │
│ │
│ │
│ 🖥️ Server name: ComputeControl │
│ 📦 Transport: STDIO │
│ │
│ 📚 Docs: https://gofastmcp.com │
│ 🚀 Deploy: https://fastmcp.cloud │
│ │
│ 🏎️ FastMCP version: 2.10.6 │
│ 🤝 MCP version: 1.12.0 │
│ │
╰────────────────────────────────────────────────────────────────────────────╯
[08/23/25 08:21:49] INFO Starting MCP server 'ComputeControl' with transport 'stdio' server.py:1371
response_type: <class 'fastmcp.utilities.json_schema_type.UserInput'>
panda のステータスは STOPPED です. 起動しますか? [y/N]:
MCP Server から Elicitation のコールバックが入りました。
y を入力して処理を継続します。
panda のステータスは STOPPED です. 起動しますか? [y/N]: y
result: action='accept' data=UserInput(value='y')
panda を起動しました、ステータスは "STARTING" です
result: action='accept' data=UserInput(value='y') の行は MCP Server が ctx.elicit() の返り値を出力しているものです。入力値 y が UserInput クラスのオブジェクトとして返されているのが分かります。
VS Code GitHub Copilot Chat で実行する
では、今度はいよいよ自然言語を使って指示を出してみます。
MCP Server を登録する ( .vscode/mcp.json を作成)
VS Code のワークスペースで使用可能な MCP Server は、.vscode/mcp.json で管理されます。まず、.vscode/mcp.json を新規作成してエディタで開くと、右下に自動的にボタンが表示されます。
Add Server... をクリックすると、MCP Server のタイプを聞いてきますので、Command (stdio) を選択します。
そうすると、mcp.json ファイルに定義ファイルの雛形が出来ますので、これに情報を追加していきます。定義ファイルのフォーマットは こちら で参照できます。${userHome} や ${workspaceFolder} のような変数も利用できます。
編集が完了したら、下図の矢印で示しているところにある Start をクリックして MCP Server を開始させて下さい。
Runnung になっていることを確認。
自然言語を使ってインスタンスを起動する
Copilot Chat の Configure Tools （下図の矢印）をクリックして、登録した MCP Server が有効かどうか確認しましょう。
compute_control という MCP Server が有効になっていることを確認して下さい。
では、自然言語で指示を出してみます。
「インスタンス panda を起動して」
すると、最初に list() が呼ばれ、次に action() が呼ばれたところで、MCP Server から Elicitation の仕組みを使ってユーザーからの入力を求めるリクエストが入りました。
panda のステータスは STOPPED です. 起動しますか? [y/N]
Respond をクリックすると、上部に入力フィードが現れますので、そこに y と入力します。
MCP Server はユーザーからの入力 y を受け取り panda インスタンスを起動します。
うまくいきましたね。実際にLLMが実行した内容は > Ran list、> Ran action の部分を展開して確認することができます。
インスタンスの停止も同じように自然言語で指示できますので試してみて下さい。
まとめ
MCP の Elicitation を使って、MCP Server とインタラクティブなやりとりを行いました。実行の確認/キャンセルだけでなく、実行に必要な情報の更新や追加を行うことができます。今回は非常に単純な単一の str 型のデータしか扱いませんでしたが、複数項目＆タイプの情報をユーザーに入力してもらうことも可能です。