こんにちは、JS2IIUです。
最近よく耳にするMCP、すでにさまざまなサービスが提供され始めています。MCPについて理解を深めるため、Streamlitと組み合わせて実装してみたいと思います。今回もよろしくお願いします。
1. はじめに
近年、ChatGPTやClaudeなどの大規模言語モデル（LLM）が驚くべきスピードで進化しています。しかし、どれだけ言語生成が賢くなっても、それだけでは実用的なアプリケーションにはなりません。モデルが社内データにアクセスしたり、タスク管理ツールに書き込んだりと、外部ツールと安全かつ柔軟に接続できる仕組みが必要になります。
そこで登場するのが MCP（Model Context Protocol） です。
本記事では、MCPの基本概念を丁寧に解説しつつ、実際にMCPサーバーを構築してStreamlitでUIから呼び出す体験を通じて、プロトコルの仕組みを深く理解していきます。
MCPについてはこちらの関連記事でも詳しく説明しています。
2. MCPとは何か？仕組みをやさしく解説
2.1 MCPの定義と役割
MCP（Model Context Protocol） とは、LLMと外部ツールやデータソースを安全・汎用的に接続するためのプロトコルです。OpenAIが提唱するこのプロトコルは、従来のFunction Callingやプラグインの複雑さを抽象化し、より構造化された方法で外部機能とやり取りできるように設計されています。
- LLMは「MCPクライアント」を通じて「MCPサーバー」にリクエストを送信。
- MCPサーバーは関数やデータを提供する。
- すべてのやりとりはJSONベースの構造化された通信で行われる。
2.2 MCPが使われるシナリオ例
| シナリオ | 説明 |
|---|---|
| Googleカレンダー操作 | LLMが予定を取得・登録 |
| Notion連携 | LLMがページに追記・検索 |
| 自社DB接続 | 顧客情報や販売履歴を照会 |
| ローカルスクリプト呼び出し | ユーザー環境のスクリプト実行 |
3. MCPの3つの要素とは？
MCP（Model Context Protocol）は、大規模言語モデル（LLM）が外部リソース（ツール、API、データベースなど）と安全かつ一貫性のある方法で接続できるように設計された通信プロトコルです。
このプロトコルは、次の3つのコンポーネントから構成されており、それぞれが異なる責任範囲を担っています。これらの構成要素の役割と関係性を正しく理解することで、MCPベースのアーキテクチャを効果的に設計・実装できるようになります。
3.1 MCPホスト（Model Host）
- 役割：LLMアプリケーションの「本体」であり、ユーザーの入力を受け取って解釈し、必要に応じて外部リソースへのアクセスを指示する中枢です。
- 例：ChatGPT、Claude、Anthropic Consoleなど
- 主な機能：
- ユーザーからの自然言語プロンプトを解析し、意図された操作（ツールの呼び出し）を判断
- MCPクライアントの生成・管理（リクエストごとにクライアントを起動する場合もあれば、長期接続で維持する設計も可能）
- 出力結果をユーザーに返す
技術的には、ホストはLLMが生成した中間表現（例：ToolCall）をMCPクライアントに転送し、関数の呼び出し処理を委譲します。
3.2 MCPクライアント（Model Client）
- 役割：MCPホストとMCPサーバーの仲介役となるプロセス
- 構造：1つのMCPクライアントは、1つのMCPサーバーと1対1で接続し、セッション単位で動作します
- 主な機能：
- ホストから受け取ったツール呼び出しリクエスト（関数名・引数）をMCPサーバーにHTTPリクエストとして転送
- MCPサーバーからのレスポンスを受け取り、ホストに返却
- 接続のライフサイクル（起動・再接続・シャットダウン）管理
MCPクライアントは通常、OpenAIが提供するインフラ上で動作しますが、開発者が自前でクライアントを模倣・テストすることも可能です。HTTPベースの単純なプロトコルであるため、Postmanやcurlなどを使った手動テストもできます。
3.3 MCPサーバー（Tool Server）
- 役割：実際の処理ロジックを実装するエンドポイント群で、外部の業務ロジック、ツール、データソースへのアクセスを提供します
- 実装形態：RESTfulなWebサービス（FastAPI, Flask, Express.jsなどで実装可能）
- 主な機能：
- ユーザーから指定された関数名と入力引数を受け取り、適切な処理を実行
- 実行結果をMCPクライアントに返却（JSON形式）
- 処理単位でログ記録やセキュリティ制御を実装可能
MCP仕様上、サーバーは以下の2つのエンドポイントを必須で実装しなければなりません：
| エンドポイント | 概要 |
|---|---|
GET /health | サーバーが正常に動作しているかを返すステータス確認用API |
POST /call | 呼び出したい関数名と引数を含むリクエストを受け取り、処理結果を返す主エンドポイント |
技術的には、
/callは関数名（例：greet,search_customer,add_task）と引数（inputs）を含むJSONを受け取り、同様にJSON形式でoutputsを返す必要があります。MCPホストはこの形式に従ってのみツールとの通信を行うため、仕様に準拠して実装することが重要です。
図で理解する：MCP全体の構成イメージ
このように、MCPはLLMと外部ツールを安全・構造的に連携させるための明確な分離構造を提供します。どこにどの処理責任があるのかを明確にすることで、セキュリティ、スケーラビリティ、保守性が高まります。
次章では、これらの構成のうち「MCPサーバー」を実際に実装し、その動作を確認してみましょう。
4. MCPサーバーを作ってみよう（FastAPI + Streamlit）
ここからは、MCPサーバーの実装を実際に体験しながら学んでいきましょう。MCPサーバーは、MCPクライアントからのリクエストを受け取り、関数を呼び出してその結果を返すREST APIです。
今回は、Python製の高速なWebフレームワークである FastAPI を使ってMCPサーバーを実装し、Streamlitを使ってUIからAPIを操作できるようにします。
4.1 開発環境の準備
MCPサーバーの構築には、以下のPythonライブラリを使用します。
| ライブラリ | 用途 |
|---|---|
fastapi | Web API の実装 |
uvicorn | ASGIサーバー。FastAPIを起動するために必要 |
pydantic | リクエスト／レスポンスのデータバリデーション |
streamlit | ユーザーインターフェース用（後半で使用） |
仮想環境を作成（推奨）
Pythonの仮想環境を作って、プロジェクトごとに依存関係を分離しましょう。
python -m venv .venv
source .venv/bin/activate # Windows の場合: .venv\Scripts\activateライブラリをインストール
pip install fastapi uvicorn streamlit pydantic4.2 MCPサーバーの基本実装
FastAPIでMCPサーバーを作るには、最低限2つのエンドポイントを実装する必要があります。
GET /health：MCPホストがサーバーの正常性を確認するためのチェック用POST /call：ツール（関数）を呼び出すためのメイン処理
4.2.1 /health エンドポイントの実装
このエンドポイントは、MCPホストがサーバーと接続できるかどうかを確認するために利用されます。とてもシンプルな構造で、サーバーが稼働中であることを返します。
# app.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/health")
def health_check():
return {"status": "ok"}解説
FastAPI()：FastAPIアプリケーションのインスタンスを作成します。@app.get("/health")：GET /healthに対応するエンドポイントを定義。- 関数
health_checkは辞書型（JSON）で"status": "ok"を返します。
実運用では、ヘルスチェックにCPU使用率やDB接続状態を加えることもありますが、MCP準拠のサーバーではこの簡単な形式で十分です。
4.2.2 /call エンドポイントの実装
このエンドポイントは、MCPクライアントから送信される「関数名」と「引数」を受け取り、指定された処理を実行し、出力結果をJSON形式で返す役割を担います。
from pydantic import BaseModel
from typing import Dict
class CallRequest(BaseModel):
function: str
inputs: Dict[str, str]
class CallResponse(BaseModel):
outputs: Dict[str, str]
@app.post("/call", response_model=CallResponse)
def call(req: CallRequest):
if req.function == "greet":
name = req.inputs.get("name", "world")
return {"outputs": {"message": f"Hello, {name}!"}}
else:
return {"outputs": {"message": "Unknown function"}}解説
データモデルの定義（Pydantic）
class CallRequest(BaseModel):
function: str
inputs: Dict[str, str]function：呼び出したい関数名（文字列）inputs：関数に渡される引数を格納した辞書（key-value）
class CallResponse(BaseModel):
outputs: Dict[str, str]outputs：関数実行後に返す結果（辞書形式）
MCP仕様では、リクエストは必ず
functionとinputsを持ち、レスポンスは必ずoutputsを持つ必要があります。
エンドポイントのロジック
@app.post("/call", response_model=CallResponse)
def call(req: CallRequest):@app.post("/call")：HTTPのPOSTリクエストを/callに送るとこの関数が実行されます。response_model=CallResponse：出力JSONをPydanticモデルでバリデーション・整形します。
if req.function == "greet":
name = req.inputs.get("name", "world")
return {"outputs": {"message": f"Hello, {name}!"}}function == "greet"のとき、inputsからnameを取得し、挨拶メッセージを返します。inputs.get("name", "world")：指定がなければデフォルトで “world” を使います。
else:
return {"outputs": {"message": "Unknown function"}}- 未定義の関数名が来た場合は、”Unknown function” というメッセージで応答します。
動作確認例（curl）
以下のコマンドを使って、/call エンドポイントをテストできます。
curl -X POST http://localhost:8000/call \
-H "Content-Type: application/json" \
-d '{"function": "greet", "inputs": {"name": "Makoto"}}'期待されるレスポンス：
{
"outputs": {
"message": "Hello, Makoto!"
}
}このように、FastAPIを使えば、MCP準拠のサーバーを非常に少ないコード量で実装することができます。次のセクションでは、このMCPサーバーにStreamlitから接続して、UIを使って動かしてみます。
5. StreamlitからMCPサーバーを呼び出してみよう
MCPクライアントのような役割を果たすUIをStreamlitで作成し、サーバーにデータを送ってみましょう。
5.1 Streamlitアプリのコード
# client.py
import streamlit as st
import requests
st.title("MCP クライアント シミュレーター")
name = st.text_input("名前を入力してください")
if st.button("送信"):
res = requests.post("http://localhost:8000/call", json={
"function": "greet",
"inputs": {"name": name}
})
if res.status_code == 200:
st.success(res.json()["outputs"]["message"])
else:
st.error("エラーが発生しました")5.2 サーバーの起動コマンド
uvicorn app:app --reload別ターミナルで以下を実行してUIを表示：
streamlit run client.py6. 発展：複数の関数をMCPサーバーに追加してみよう
MCPは「複数の関数」に対応できます。たとえば「add」関数を追加してみましょう。
6.1 足し算関数の追加
def add(x: int, y: int) -> int:
return x + y
@app.post("/call", response_model=CallResponse)
def call(req: CallRequest):
if req.function == "greet":
name = req.inputs.get("name", "world")
return {"outputs": {"message": f"Hello, {name}!"}}
elif req.function == "add":
try:
x = int(req.inputs.get("x", "0"))
y = int(req.inputs.get("y", "0"))
return {"outputs": {"result": str(add(x, y))}}
except Exception as e:
return {"outputs": {"error": str(e)}}
else:
return {"outputs": {"message": "Unknown function"}}Streamlit 側も以下のように修正することで、複数関数を選択可能にできます。
function = st.selectbox("呼び出す関数", ["greet", "add"])
if function == "add":
x = st.number_input("x", value=0)
y = st.number_input("y", value=0)
inputs = {"name": name} if function == "greet" else {"x": str(x), "y": str(y)}
if st.button("送信"):
res = requests.post("http://localhost:8000/call", json={
"function": function,
"inputs": inputs
})
...7. 参考リンク・ドキュメント
最後まで読んでいただきありがとうございます。
ご意見ご感想はコメント欄にお願い致します。
コメント