Azure Functions の MCP 拡張機能
Azure Functions の MCP 拡張機能プレビュー版が 2025/04/05 に発表されました。
MCP 拡張機能によって、自作の MCP サーバーを Azure Functions (リモート) で公開できたり、バインディング機能を使って Azure 製品を掛け合わせた MCP サーバーの構築ができたりと、多くのメリットがあるかと思います。
Azure Functions の MCP ツール実装は以下の記事にも紹介がありますので、より詳しい内容はこちらを参照ください。
Azure Container Apps Dynamic Sessions とは？
Azure Container Apps Dynamic Sessions は、Azure Container Apps 上で動的にコードを実行できるサービスです。
外部から送信された Python スクリプトなどを、セッション単位で安全に実行できるため、LLM が生成した信頼されていないコードもサンドボックス内でセキュアに実行できます。
背景
Microsoft Azure Developers チャンネルに Dynamic Sessions Code Interpreter を AI フレームワークと統合し、Agent が生成したコードを実行するデモ動画がありました。
この実装では LangChain や Semantic Kernel など別途 AI フレームワークを用意して、Dynamic Sessions との統合を実現しています。動画のコードではないですが、簡易的なサンプルは以下リンクにて公開されています。
フレームワークに依存しない実装で、MCP による標準化が出来れば、より柔軟性・拡張性のあるツールとして活用できるのではないかと思い、今回の構成を考えるに至りました。
TL;DR
- Dynamic Sessions を Rest API 経由で呼び出し、Python コードを実行する関数
execute_dynamic_sessionを作成。 - MCP トリガー関数
run_dynamic_sessionを作成し、MCP ツール化。 - VSCode の Copilot Agent から MCP ツールを呼び出し。
- Agent が生成した Python コードを MCP ツールから実行する流れを検証。
実装手順
Dynamic Sessions リソースの作成と権限付与
こちらの手順は少し長くなるため、割愛させていただきます🙇
以下チュートリアルの「コード実行 API のロールの割り当てを設定する」までを実施します。
このチュートリアルでは JavaScript コードの実行ですが、今回は Python コードを実行させたいので、リソース作成時のコマンドは --container-type を PythonLTS とします。
Azure Functions MCP トリガーの実装準備
今回はローカル MCP サーバーを作成しますが、将来的にはリモート化もできればと思い、以下サンプルを使用しました。
README の手順に沿って hello_mcp ツールの動作を確認してみると、Agent がローカル MCP ツールを認識し、関数 hello_mcp を呼び出していることが分かります。
関数 hello_mcp の実装は以下の通りです。現時点では generic decorator を使用して MCP トリガーを定義する必要があります。
@app.generic_trigger(
arg_name="context",
type="mcpToolTrigger",
toolName="hello_mcp",
description="Hello world.",
toolProperties="[]",
)
def hello_mcp(context) -> None:
"""
A simple function that returns a greeting message.
Args:
context: The trigger context (not used in this function).
Returns:
str: A greeting message.
"""
return "Hello I am MCPTool!"
decorator の各プロパティについては以下ドキュメントに説明がありますが、特に toolProperties は MCP ツールの期待する入力を設定できる個所になるため、ここの設計は重要になると思います。
arg_name から関数が Agent のコンテキストを取得することができますが、関数 hello_mcp は toolProperties が空となっており、そもそもコンテキストの受け取りを想定していない実装となっています。
Dynamic Sessions 呼び出し関数の作成
Dynamic Session の Rest API を Python コードから実行する関数 execute_dynamic_session を作成しました。
from azure.identity import DefaultAzureCredential
import requests
... (省略) ...
def execute_dynamic_session(code: str, identifier: str = "mcp-test") -> dict:
if not code:
return {"error": "code プロパティが指定されていません。"}
base = os.environ.get("SESSION_POOL_CODE_EXECUTE_ENDPOINT")
api_ver = os.environ.get("DYNAMIC_SESSIONS_API_VERSION", "2024-02-02-preview")
if not base:
return {"error": "環境変数 SESSION_POOL_CODE_EXECUTE_ENDPOINT が未設定です。"}
url = f"{base}?api-version={api_ver}&identifier={identifier}"
logging.info(f"Dynamic Sessions request URL: {url}")
# Azure 認証
try:
credential = DefaultAzureCredential()
token = credential.get_token("https://dynamicsessions.io/.default")
access_token = token.token
except Exception as ex:
logging.exception("Azure 認証に失敗しました。")
return {"error": f"認証エラー: {str(ex)}"}
# API 呼び出し
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
}
payload = {
"properties": {
"codeInputType": "inline",
"executionType": "synchronous",
"code": code,
"timeoutInSeconds": 100
}
}
logging.info(f"Request payload: {json.dumps(payload, ensure_ascii=False)}")
try:
resp = requests.post(url, headers=headers, json=payload, timeout=120)
resp.raise_for_status()
logging.info(f"Dynamic Sessions API Response: {resp.status_code}")
logging.info(f"Dynamic Sessions API Response Headers: {resp.headers}")
logging.info(f"Dynamic Sessions API Response Body: {resp.text}")
return resp.json()
except requests.exceptions.RequestException as ex:
logging.exception("Dynamic Sessions 呼び出しに失敗しました。")
error_details = {
"error": f"API呼び出しエラー: {str(ex)}",
"status_code": getattr(ex.response, 'status_code', None),
"response_text": getattr(ex.response, 'text', None),
"url": url
}
if hasattr(ex, 'response') and ex.response is not None:
try:
error_details["response_json"] = ex.response.json()
except json.JSONDecodeError:
pass
logging.error(f"エラー詳細: {json.dumps(error_details, ensure_ascii=False, indent=2)}")
return error_details
except Exception as ex:
logging.exception("予期しないエラーが発生しました。")
return {"error": f"内部エラー: {str(ex)}"}
- 環境変数は local.settings.json に記載。
- アクセストークンの取得では
DefaultAzureCredentialを使用しているので、requirements.txt にazure-identityの追記が必要。 - DYNAMIC_SESSIONS_API_VERSION は 2024-02-02-preview を使用。
- SESSION_POOL_CODE_EXECUTE_ENDPOINT は以下コマンドで取得した値。
az containerapp sessionpool show -n $SESSION_POOL_NAME -g $RESOURCE_GROUP --query "properties.poolManagementEndpoint" -o tsv
他の API バージョンを使用する場合は、適宜 API バージョンに沿ったエンドポイント・プロパティの指定が必要になります。
テスト用に HTTP トリガー関数の作成
関数 execute_dynamic_session の動作確認をしたかったので HTTP トリガー関数を作成しました。
テスト用なので、MCP ツールの作成には不要です。
@app.route(route="test_dynamic_session", methods=["POST"])
def test_dynamic_session(req: func.HttpRequest) -> func.HttpResponse:
try:
req_body = req.get_json()
if not req_body:
return func.HttpResponse(
json.dumps({"error": "JSON body が必要です。"}),
status_code=400,
mimetype="application/json"
)
code = req_body.get("code")
identifier = req_body.get("identifier", "test")
if not code:
return func.HttpResponse(
json.dumps({"error": "code フィールドが必要です。"}),
status_code=400,
mimetype="application/json"
)
result = execute_dynamic_session(code, identifier)
return func.HttpResponse(
json.dumps(result, ensure_ascii=False, indent=2),
status_code=200,
mimetype="application/json"
)
except Exception as ex:
logging.exception("テスト用エンドポイントでエラーが発生しました。")
return func.HttpResponse(
json.dumps({"error": f"Internal error: {str(ex)}"}),
status_code=500,
mimetype="application/json"
)
HTTP トリガー関数内で関数 execute_dynamic_session を呼び出し、Dynamic Sessions 内のコード実行をテストします。
テスト用 HTTP トリガー関数の実行
VS Code の拡張ツール (REST Client) を使用して HTTP トリガー関数を呼び出します。
※http_auth_level=func.AuthLevel.FUNCTION なので、Azure 上で動作確認をする際は functionKey を指定する必要があります。
@baseUrl = http://localhost:7071
@functionKey =
POST {{baseUrl}}/api/test_dynamic_session?code={{functionKey}}
Content-Type: application/json
{
"code": "import math\nimport json\n\nresult = {\n 'pi': math.pi,\n 'sqrt_16': math.sqrt(16),\n 'factorial_5': math.factorial(5)\n}\nprint(json.dumps(result, indent=2))",
"identifier": "test-math"
}
code プロパティで渡した Python コードの実行結果を取得できたことが分かります。
{
"$id": "1",
"properties": {
"$id": "2",
"status": "Success",
"stdout": "{\n \"pi\": 3.141592653589793,\n \"sqrt_16\": 4.0,\n \"factorial_5\": 120\n}\n",
"stderr": "",
"result": "",
"executionTimeInMilliseconds": 58
}
}
これで無事 Dynamic Session の呼び出し関数 execute_dynamic_session が正常に動作していることを確認できたので、MCP トリガー関数の作成に移ろうと思います。
MCP トリガー関数の作成
Dynamic Session 呼び出し関数を MCP トリガー関数内で実行するために、Agent から期待する入力を設定します。
class ToolProperty:
def __init__(self, property_name: str, property_type: str, description: str):
self.propertyName = property_name
self.propertyType = property_type
self.description = description
def to_dict(self):
return {
"propertyName": self.propertyName,
"propertyType": self.propertyType,
"description": self.description,
}
tool_props_run_ds_obj = [
ToolProperty("code", "string", "実行する Python コード文字列 (必須)"),
ToolProperty("identifier", "string", "Dynamic Session の識別子 (省略可)")
]
tool_props_run_ds_json = json.dumps([p.to_dict() for p in tool_props_run_ds_obj])
-
codeプロパティ：実行する Python コードの文字列です。Dynamic Sessions へ渡すコードになるため、必須と表記しています。 -
identifierプロパティ：コードをセッションプールのどのセッションで実行するか指定する識別子になります。詳しくは以下ドキュメントに紹介があります。
なお、description に (必須)、(省略可) と書いてますが、Agent のコンテキストを完全に制御はできません。ただ、"プロンプト次第" でコントロールは可能です。
※何度か検証し、code プロパティが指定されないことはなかったですが、identifier プロパティの description に (必須) と書いても、identifier プロパティを指定してくれないことがありました。
プロンプトに「identifier は hogehoge としてください」と入力することで、コンテキストに "identifier": "hogehoge" が追加されました。特定の identifier で実行したい場合はプロンプトから指定するのがよさそうです。
{
"code": "import subprocess\nimport sys\nimport json\n\nlocation = \"London\"\nidentifier = \"hogehoge\"\nresult = {\"method\": None, \"success\": False, \"output\": None}\n\n# Helper to try subprocess commands\ndef try_cmd(cmd):\n try:\n p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=15)\n return p.returncode, p.stdout.strip(), p.stderr.strip()\n except FileNotFoundError:\n return None, None, None\n except Exception as e:\n return 1, None, str(e)\n\n# 1) Try common mcp CLI invocation patterns\ncmd_attempts = [\n [\"mcp\", \"call\", \"weather\", \"--location\", location, \"--identifier\", identifier],\n [\"mcp\", \"weather\", \"get\", \"--city\", location, \"--id\", identifier],\n [\"mcp\", \"weather\", location, \"--id\", identifier],\n [\"mcp\", \"weather\", \"--location\", location, \"--identifier\", identifier],\n]\n\nfor cmd in cmd_attempts:\n rc, out, err = try_cmd(cmd)\n if rc is None:\n # command not found\n continue\n if rc == 0 and out:\n result.update({\"method\": \"mcp_cli\", \"success\": True, \"cmd\": cmd, \"output\": out})\n print(json.dumps(result))\n sys.exit(0)\n # record last non-zero attempt if any\n if out or err:\n result.update({\"last_attempt\": {\"cmd\": cmd, \"returncode\": rc, \"stdout\": out, \"stderr\": err}})\n\n# 2) Try importing a Python mcp module and invoking common call patterns\ntry:\n import mcp as mcp_mod\n # try several call patterns\n tried = {}\n for attr in (\"call\", \"invoke\", \"run\", \"request\"):\n fn = getattr(mcp_mod, attr, None)\n if callable(fn):\n try:\n out = fn(\"weather\", location=location, identifier=identifier)\n result.update({\"method\": \"mcp_python\", \"success\": True, \"func\": attr, \"output\": out})\n print(json.dumps(result))\n sys.exit(0)\n except TypeError:\n try:\n out = fn({\"action\": \"weather\", \"location\": location, \"identifier\": identifier})\n result.update({\"method\": \"mcp_python\", \"success\": True, \"func\": attr, \"output\": out})\n print(json.dumps(result))\n sys.exit(0)\n except Exception as e:\n tried[attr] = str(e)\n except Exception as e:\n tried[attr] = str(e)\n result[\"mcp_python_tried\"] = tried\nexcept Exception as e:\n result[\"mcp_python_error\"] = str(e)\n\n# 3) Fallback: use wttr.in (simple public CLI-style weather)\ntry:\n import requests\n url = f\"https://wttr.in/{location}?format=j1\"\n r = requests.get(url, timeout=10)\n if r.status_code == 200:\n data = r.json()\n # create a concise summary\n current = data.get('current_condition', [{}])[0]\n temp_C = current.get('temp_C')\n weather_desc = current.get('weatherDesc', [{}])[0].get('value') if current.get('weatherDesc') else None\n feels_like = current.get('FeelsLikeC')\n summary = f\"{location}: {weather_desc}, {temp_C}°C (feels like {feels_like}°C)\"\n result.update({\"method\": \"wttr.in\", \"success\": True, \"output\": summary, \"raw\": data})\n print(json.dumps(result))\n sys.exit(0)\n else:\n result.update({\"method\": \"wttr.in\", \"success\": False, \"status_code\": r.status_code})\nexcept Exception as e:\n result.update({\"method\": \"wttr.in\", \"success\": False, \"error\": str(e)})\n\n# If we reach here, nothing succeeded\nprint(json.dumps(result))\nsys.exit(0)\n",
"identifier": "hogehoge"
}
今回はコードロジックで identifier プロパティがない場合の処理を書くことで、エラーを回避しています。
MCP トリガー関数の実装は以下の通りです。上記で作成した json を toolProperties に渡しています。関数 run_dynamic_session は Agent のコンテキスト (context) へアクセスし、code や identifier を取得しています。
@app.generic_trigger(
arg_name="context", type="mcpToolTrigger",
toolName="run_dynamic_session",
description="Azure Container Apps Dynamic Sessions に Python コードを渡して実行します。",
toolProperties=tool_props_run_ds_json,
)
def run_dynamic_session(context) -> str:
args = json.loads(context).get("arguments", {}) or {}
code = args.get("code")
identifier = args.get("identifier") or "mcp-test"
execute_result = execute_dynamic_session(code, identifier)
logging.info(f"Row Mcp result: {execute_result}")
return execute_result
動作確認
作成した MCP ツールを使用して、現在のロンドンの天気を取得してみます。
実際に LLM が生成したコンテキストと Python コードは以下の通りです。
urllib.request で 天気予報 API (Open-Meteo) を叩き、天気データを取得する実装となっています。
{
"code": "import urllib.request\nimport json\n\nurl = \"https://api.open-meteo.com/v1/forecast?latitude=51.5072&longitude=-0.1276¤t_weather=true\"\n\nweather_code_map = {\n 0: \"快晴\",\n 1: \"主に晴れ\",\n 2: \"一部曇り\",\n 3: \"曇り\",\n 45: \"霧\",\n 48: \"霧（凍結性）\",\n 51: \"霧雨（弱）\",\n 53: \"霧雨（中等）\",\n 55: \"霧雨（強）\",\n 56: \"凍結霧雨（弱）\",\n 57: \"凍結霧雨（強）\",\n 61: \"雨（弱）\",\n 63: \"雨（中等）\",\n 65: \"雨（強）\",\n 66: \"凍結雨（弱）\",\n 67: \"凍結雨（強）\",\n 71: \"雪（弱）\",\n 73: \"雪（中等）\",\n 75: \"雪（強）\",\n 77: \"雪粒\",\n 80: \"にわか雨（弱）\",\n 81: \"にわか雨（中等）\",\n 82: \"にわか雨（強）\",\n 85: \"にわか雪（弱）\",\n 86: \"にわか雪（強）\",\n 95: \"雷雨（軽〜中等）\",\n 96: \"雷雨（軽い雹を伴う）\",\n 99: \"雷雨（激しい雹を伴う）\",\n}\n\ntry:\n with urllib.request.urlopen(url, timeout=15) as r:\n data = json.load(r)\nexcept Exception as e:\n print(json.dumps({\"ok\": False, \"error\": str(e)}))\nelse:\n cw = data.get(\"current_weather\") or {}\n if not cw:\n print(json.dumps({\"ok\": False, \"error\": \"current_weather が取得できませんでした\", \"raw\": data}))\n else:\n temp = cw.get(\"temperature\")\n windspeed = cw.get(\"windspeed\")\n winddir = cw.get(\"winddirection\")\n code = cw.get(\"weathercode\")\n time = cw.get(\"time\")\n desc = weather_code_map.get(code, f\"天気コード {code}\")\n result = {\n \"ok\": True,\n \"location\": \"London\",\n \"time\": time,\n \"temperature_c\": temp,\n \"windspeed_m_s\": windspeed,\n \"winddirection_deg\": winddir,\n \"weathercode\": code,\n \"description_ja\": desc,\n \"source\": \"open-meteo.com\"\n }\n print(json.dumps(result, ensure_ascii=False))\n",
"identifier": "get_london_weather"
}
LLM が生成した Python コード
import urllib.request
import json
url = "https://api.open-meteo.com/v1/forecast?latitude=51.5072&longitude=-0.1276¤t_weather=true"
weather_code_map = {
0: "快晴",
1: "主に晴れ",
2: "一部曇り",
3: "曇り",
45: "霧",
48: "霧（凍結性）",
51: "霧雨（弱）",
53: "霧雨（中等）",
55: "霧雨（強）",
56: "凍結霧雨（弱）",
57: "凍結霧雨（強）",
61: "雨（弱）",
63: "雨（中等）",
65: "雨（強）",
66: "凍結雨（弱）",
67: "凍結雨（強）",
71: "雪（弱）",
73: "雪（中等）",
75: "雪（強）",
77: "雪粒",
80: "にわか雨（弱）",
81: "にわか雨（中等）",
82: "にわか雨（強）",
85: "にわか雪（弱）",
86: "にわか雪（強）",
95: "雷雨（軽〜中等）",
96: "雷雨（軽い雹を伴う）",
99: "雷雨（激しい雹を伴う）",
}
try:
with urllib.request.urlopen(url, timeout=15) as r:
data = json.load(r)
except Exception as e:
print(json.dumps({"ok": False, "error": str(e)}))
else:
cw = data.get("current_weather") or {}
if not cw:
print(json.dumps({
"ok": False,
"error": "current_weather が取得できませんでした",
"raw": data
}))
else:
temp = cw.get("temperature")
windspeed = cw.get("windspeed")
winddir = cw.get("winddirection")
code = cw.get("weathercode")
time = cw.get("time")
desc = weather_code_map.get(code, f"天気コード {code}")
result = {
"ok": True,
"location": "London",
"time": time,
"temperature_c": temp,
"windspeed_m_s": windspeed,
"winddirection_deg": winddir,
"weathercode": code,
"description_ja": desc,
"source": "open-meteo.com"
}
print(json.dumps(result, ensure_ascii=False))
Agent のコンテキストを MCP トリガーが認識し、関数内の処理で Dynamic Sessions にコードを渡します。Dynamic Sessions が返した実行結果 (標準出力・標準エラー) をそのまま Agent へ返しており、無事ロンドンの天気情報を取得できたことが分かります。
このように、ローカル環境が閉域化されている状況でも、Dynamic Sessions を介して Azure のサンドボックス環境から外部 API を安全に呼び出すことが可能です。
まとめ
AI フレームワークを介さず、直接 Dynamic Sessions の REST API を呼び出す関数を作成し、Azure Functions の MCP 拡張機能を活用することで、手軽に MCP ツールを構築できました。
Dynamic Sessions はコード実行だけでなく、ファイル操作用のエンドポイントも備えており、様々なシナリオでの活用が期待できそうです。
補足
Azure Functions の MCP 拡張機能は 2025/10/27 現在、GitHub に一般公開版 Release されていましたが、公式のアナウンスはまだでした。
Azure Container Apps Dynamic Sessions は Python コードインタープリターが GA されていますが、JavaScript コードインタープリターはパブリックプレビュー段階になります。
Discussion