MCPサーバーの作り方 — PythonとFastMCPでデータベースにAIが日本語で答えるようにする（前編）
MCPサーバーの作り方を、動くコードつきで最初から解説します。PythonとFastMCPでSQLiteの売上データベースをMCP化し、AIに日本語で聞くとAIが自分でSQLを書いて集計まで返すところまで作ります。
こんにちは！Qualitegプロダクト開発部です！
手元のデータベースをAIから問い合わせられるようにしたい。 そのために必要なのがMCPサーバーです。
やりたいことは単純なのですが、いざ調べてみると「MCPサーバーを作ればいい」というところまでは分かっても、自分のデータベースをどうやってMCP化するのかでつまずきます。
この記事では、
PythonとFastMCPでMCPサーバーを自作し、CLI版のClaude Codeから売上データベースに日本語で聞けるようにする
までの手順を、動くコードつきで最初から最後まで書きます。コードは200行ほどです。「職業別の売上トップ3を教えて」と聞くと、Claude Codeが自分でSQLを組み立てて集計まで返してくれるようになります。
書いたコードは全部GitHubに置きました。cloneして、依存関係のインストールとDB作成を行えば動きます。動かす手順はリポジトリのREADMEにも全部書いてあるので、記事を読まずに手を動かしたい方はそちらからどうぞ。
qualiteg/mcp-server-tutorial ｜ サンプルコード一式とREADME（GitHub）
前後編に分かれていて、前編（この記事）ではCLI版のClaude Codeから使えるところまで。後編では、同じサーバーをWeb版のChatGPT・Claudeから使えるようにします。手元で動くCLI版と、ブラウザで動くWeb版。この2つのあいだに、多くの人が引っかかる壁があります。
MCPとは何か、3分で
MCP（Model Context Protocol）は、AIアプリと外部のデータやツールをつなぐための共通規格です。Anthropicが2024年末に公開し、いまではClaudeだけでなくChatGPT、Gemini、各種コーディングエージェントが対応しています。
MCPが無かったころは、AIにデータベースを触らせたければ、AIアプリごとの独自プラグインや関数呼び出しに個別対応する必要がありました。MCPでは「サーバー（データやツールを出す側。今回作るもの）」と「クライアント（ClaudeやChatGPTなどのAIアプリ側）」の話し方が規格で決まっています。一度MCPサーバーを作れば、AIアプリごとに専用プラグインを作り直さずに、MCP対応の複数のAIアプリから利用できます。ただし、使えるトランスポートや認証方式、プラン、管理者設定はクライアントごとに違います。
MCPサーバーが出せるものはtools・resources・promptsの3種類ありますが、実務でまず使うのはtools、つまりAIが呼び出せる関数です。この記事もtoolsだけ扱います。
もうひとつ、後編で効いてくる予備知識があります。MCP仕様が定める標準トランスポートは2系統あります。
| トランスポート | 動き方 | 主な用途 |
|---|---|---|
| stdio | クライアントが手元でサーバープロセスを起動して標準入出力で会話する | クライアントと同じ端末で動かすローカルのMCPサーバー |
| Streamable HTTP | サーバーがHTTPサーバーとして常駐し、URLに接続してもらう | 独立して常駐させ、ネットワーク経由で使うMCPサーバー |
この記事では、前編でCLI版のClaude Codeからローカルのstdioサーバーを使い、後編でWeb版からリモートのStreamable HTTPサーバーを使います。ただし「CLI版ならstdio、Web版ならHTTP」と一対一で決まるわけではありません。たとえばClaude CodeはstdioにもリモートのHTTPにも接続できます。前編では両方の動かし方を試します。
何を作るか
PCパーツのオンラインショップを想定した売上データベース（SQLite）と、それに問い合わせるMCPサーバーです。
テーブルは3つです。顧客マスター（customers）が500人ぶん、商品マスター（products）がPCパーツ8カテゴリで72商品、そして売上明細（sales_transactions）が2024年から2026年までの4,000件。顧客には年齢や職業、年収帯といった属性を持たせてあるので、「職業別に見ると」のような分析ができます。
Claude Codeに渡すツールは、思い切って2つだけにします。テーブル構造と統計を返す get_database_stats と、SELECT文を実行する execute_sql_query です。
「売上集計ツール」「顧客分析ツール」と業務ごとに関数を並べる設計もあります。ただ、SQLを書けるのはClaude Code側なので、汎用のSQL実行ツールを1本渡すほうが応用が利きます。そのぶん安全側の手当てはサーバーの責任になりますが、これは後で書きます。
準備
MCPサーバー自体の実装に必要なのは、Python 3.10以降とFastMCPです。この記事と同じ形でClaude Codeから試すなら、Claude Codeを使える環境も別途必要です。
python -m pip install fastmcppip ではなく python -m pip を使ってください。pip だと、あとでMCPサーバーを起動するPythonと別の環境に入ってしまうことがあります。
FastMCPは、Python関数にデコレータを1つ付けるだけでMCPツールを定義できる高水準フレームワークです。クライアント機能やHTTPサーバー機能もまとめて使えます。筆者環境で動作確認したのはstandalone版の3.4.5です。
まず10行で、動くMCPサーバー
いきなり売上DBに行く前に、最小構成で全体像をつかみます。
# hello_mcp.py
from fastmcp import FastMCP
mcp = FastMCP(name="hello-server")
@mcp.tool(description="2つの数値を足し算します")
def add(a: int, b: int) -> int:
return a + b
if __name__ == "__main__":
mcp.run() # 既定はstdioトランスポートこれで完成です。関数の型ヒントとdescriptionから、AIに渡されるツール定義が自動生成されます。
なお python hello_mcp.py を実行しただけでは何も表示されません。stdioサーバーとして入力待ちになるだけです。動きを確かめたいときは、FastMCPのin-memoryクライアントから呼びます。
import asyncio
from fastmcp import Client
from hello_mcp import mcp
async def main():
async with Client(mcp) as c:
print([t.name for t in await c.list_tools()]) # ['add']
result = await c.call_tool("add", {"a": 120, "b": 5})
print(result.data) # 125
if __name__ == "__main__":
asyncio.run(main())Claude Codeへの登録は1行です（登録して使うところは後半でやります）。
claude mcp add hello-server -- python hello_mcp.pyうまく繋がらないときは、Pythonとスクリプトを絶対パスで指定してください。仮想環境を使っている場合は、その中のPythonを指す必要があります。
ここで注目してほしいのは、AIが「計算が必要だ」「addというツールがある」「呼ぼう」と判断する材料が、ツール名、description、引数名や型から生成されるinputSchemaしかないことです。なかでもdescriptionは、そのツールをどんな場面で使うのかを自然文で伝えられる中心的な項目で、これがMCPサーバー設計でいちばん効いてきます。
手順1: 売上データベースを用意する
SQLiteでテーブルを3本作ります。
db_setup.py（テーブル定義とダミーデータ生成の全文コード（GitHub））
cur.execute("""
CREATE TABLE sales_transactions (
transaction_id INTEGER PRIMARY KEY,
date TEXT NOT NULL, -- 'YYYY-MM-DD'
customer_id TEXT NOT NULL REFERENCES customers(customer_id),
product_id TEXT NOT NULL REFERENCES products(product_id),
product_name TEXT NOT NULL,
product_category TEXT NOT NULL,
quantity INTEGER NOT NULL,
unit_price INTEGER NOT NULL,
total_price INTEGER NOT NULL
)""")
cur.execute("CREATE INDEX idx_trans_date ON sales_transactions(date)")
cur.execute("CREATE INDEX idx_trans_customer ON sales_transactions(customer_id)")
cur.execute("CREATE INDEX idx_trans_category ON sales_transactions(product_category)")インデックスは張っておいてください。AIは遠慮なく全期間の集計クエリを投げてきます。
乱数は固定シードにしてあるので、誰が実行しても同じデータになります。筆者環境で流すとこうなりました。
この記事に出てくる数字は、すべてこのデータの実測値です。手元で同じ数字が出るはずです。
手順2: ツールのdescriptionにスキーマを丸ごと書く
ここがこの記事でいちばん伝えたいところです。
execute_sql_queryのdescriptionには、データベースのスキーマとクエリ例を全部書き込みます。
mcp_server_sales.py（descriptionの全文コード（GitHub））
@mcp.tool(description="""SQLクエリを実行して売上データを取得します。SELECT文のみサポートします。
【データベーススキーマ】
■ customers（顧客マスター）
- customer_id (TEXT): 顧客ID（例 'C0001'）
- age (INTEGER): 年齢
- gender (TEXT): 性別（'男性', '女性'）
- prefecture (TEXT): 都道府県
- occupation (TEXT): 職業（'ITエンジニア', '会社員', '学生', '自営業' など）
- annual_income (INTEGER): 年収（万円）
■ products（商品マスター）
- product_id (TEXT): 商品ID（例 'CPU_001'）
- product_category (TEXT): CPU, GPU, Memory, SSD, HDD, Motherboard, PowerSupply, PCCase
- tier (TEXT): 'high' / 'mid' / 'entry'
（以下略。全カラムをこの調子で書く）
■ sales_transactions（売上明細）
- date (TEXT): 売上日（YYYY-MM-DD形式）
- customer_id (TEXT): 顧客ID → customers.customer_id
- product_id (TEXT): 商品ID → products.product_id
- quantity / unit_price / total_price (INTEGER)
【データ期間】2024-01-01 〜 2026-12-31（4,000件）
【クエリ例】
- 職業別の売上:
SELECT c.occupation, SUM(t.total_price) AS sales
FROM sales_transactions t JOIN customers c ON t.customer_id = c.customer_id
GROUP BY c.occupation ORDER BY sales DESC
- ハイエンドGPUの売上:
SELECT p.product_name, SUM(t.total_price) AS sales
FROM sales_transactions t JOIN products p ON t.product_id = p.product_id
WHERE p.product_category = 'GPU' AND p.tier = 'high'
GROUP BY p.product_name ORDER BY sales DESC
""")
async def execute_sql_query(sql: str, ctx: Context) -> str:
...やたら長いと思われるかもしれません。でもAIが正しいSQLを書けるかどうかは、モデルの賢さよりもスキーマをどれだけ正確に渡せているかで大きく変わります。今回のような小規模で固定されたスキーマなら、必要なカラムとJOIN関係をdescriptionに含めておくのが有効でした。
AIがツールを選ぶ材料は、ツール名、description、引数名や型から生成されるinputSchemaです。なかでもdescriptionは、どんな場面で呼ぶツールなのかを自然文で伝えられます。
クエリ例を2〜3個入れておくのもコツです。JOINの書き方や列の使い方のお手本があると、AIが書くSQLの精度が目に見えて安定します。
手順3: SQL実行ツールに3つの安全弁を設ける
AIにSQL実行を任せる以上、書き込みの防止と、重いクエリでリソースを持っていかれることの両方に備える必要があります。役割の違う3つの安全弁を設けます。
1段目、データベースを読み取り専用で開く。 SQLiteはURI接続でmode=roを指定すると、そもそも書き込みを受け付けなくなります。あとの検査をすり抜けられても、ここで止まります。
mcp_server_sales.py（読み取り専用で接続する関数のコード（GitHub））
conn = sqlite3.connect(f"file:{DB_PATH.as_posix()}?mode=ro", uri=True)2段目、SELECT以外を弾く。 これは構文解析ではなく文字列の検査なので、明らかな書き換え命令を早い段階で落とすための保守的な補助フィルターです。最終的な書き込み防止は1段目に任せます。
mcp_server_sales.py（SELECT以外を弾く部分のコード（GitHub））
normalized = sql.strip().upper()
if not normalized.startswith("SELECT"):
return "エラー: SELECT文のみサポートしています"
for keyword in ("DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "CREATE", "TRUNCATE", "ATTACH", "PRAGMA"):
if keyword in normalized:
return f"エラー: {keyword} は許可されていません"3段目、実行時間に上限を付ける。 重い集計で固まらないよう、SQLiteの進捗ハンドラで打ち切ります。
mcp_server_sales.py（実行時間に上限を付ける部分のコード（GitHub））
import time
def install_query_timeout(conn, timeout_sec):
deadline = time.monotonic() + timeout_sec
def guard():
return 1 if time.monotonic() >= deadline else 0
# SQLite仮想マシンが1,000命令を実行するたびに guard を呼ぶ
conn.set_progress_handler(guard, 1000)ここは一度間違えました。進捗ハンドラの呼び出し回数を数えて上限にしていたのですが、set_progress_handler の第2引数は秒ではなくSQLite仮想マシンの命令数です。呼ばれた回数は実行時間そのものではないので、マシンの速度やクエリの内容で待ち時間が変わってしまいます。time.monotonic() で実時間を見るのが正解でした。
なお、この3段目は読み取り専用にするための対策ではありません。暴走したクエリでリソースを持っていかれないための対策です。それぞれ守っているものが違います。
3段が本当に効いているかどうかは、あとで実際に確かめます。
手順4: AIに返す量を、サーバー側で決める
クエリ結果を整形する部分です。返す行数に上限を付けているところを見てください。
mcp_server_sales.py（クエリ結果を整形する部分のコード（GitHub））
if not rows:
return "クエリの結果、データが見つかりませんでした。"
lines = [" | ".join(columns), "-" * 40]
for row in rows[:MAX_ROWS]: # MAX_ROWS = 50
lines.append(" | ".join("" if v is None else str(v) for v in row))
if len(rows) > MAX_ROWS:
lines.append(f"... 他 {len(rows) - MAX_ROWS} 件（全体を見るには集計するか条件を絞ってください）")
return f"クエリ結果（{len(rows)} 件）:\n\n" + "\n".join(lines)AIは平気でSELECT * FROM sales_transactionsを投げてきます。4,000行をそのまま返してもコンテキストを溢れさせるだけです。件数だけは正直に伝えて、「集計するか条件を絞って」と促すのが実用的でした。
ただしこの実装が制限しているのはAIへ返す行数だけで、SQLiteから読み込む件数は制限していません。数百万行の実データを相手にするなら、fetchmany(MAX_ROWS + 1) にする、SQL側にLIMITを付ける、クエリコストを見る、といった手当てが別に要ります。
手順5: 「まず全体像を見せる」ツールを足す
2本目のツールは、テーブル構造と統計の要約を返すだけのものです。
mcp_server_sales.py（統計情報を返すツールの全文コード（GitHub））
@mcp.tool(description="""データベースのテーブル構造・件数・データ期間・カテゴリ別売上の統計を返します。
SQLを作る前に、必ずこのツールを呼んで構造を確認してください。""")
async def get_database_stats(ctx: Context) -> str:
...descriptionの最後の一文が仕事をします。
「SQLを作る前に、必ずこのツールを呼んで」
筆者環境では、これを書いたことで、まず統計を確認してからSQLを組み立てる動きになりました。
ただしこれはモデルへのお願いであって、サーバー側で強制される制御ではありません。モデルやクライアント、会話の流れによっては、統計を見ずにいきなりSQLを投げてくることもあります。安全性をこの一文に頼らせない、というのが大事なところです。ツールの説明文は、機能の説明だけでなく、使う条件や手順をモデルへ伝える場所としても働く、くらいの理解が安全です。
使ってみる
Claude Codeに登録します。
claude mcp add sales-db -- python C:\qualiteg_examples\mcp_server_sales.py登録できたか確認します。
claude mcp listsales-db: python C:\qualiteg_examples\mcp_server_sales.py - ✔ Connected✔ Connected が出れば、AIからこのサーバーが見えています。あとは日本語で聞くだけです。
claude -p "職業別の売上トップ3を教えて"返ってきたのがこれです。
質問したのは日本語の一文だけです。AIはこちらがSQLを一行も書かないまま、テーブル構造を確認し、JOINとGROUP BYを組んだクエリを投げ、集計結果を表にまとめて返してきました。
金額はダミーデータの実測値と一致しています。さらに、質問には含めていなかった取引件数も追加で集計し、「研究者は件数が最多だが単価が低い」「経営者は件数が少ないぶん単価が高い」という読みまで付けてきました。単にSQLを生成するだけでなく、集計結果の解釈まで行っています。
「2025年に発売されたハイエンドGPUで、20代への売上が多い順に」のような多段の条件も同じ調子で通ります。SQLを書いたことがない人でも、売上データベースに話しかけられるようになりました。
ただし、自然言語で聞けることと、集計結果が業務上正しいことは別の話です。重要な判断に使うなら、実行されたSQL・集計の定義・対象期間・除外条件を確認できるようにしておいてください。「売上」「顧客数」「平均単価」あたりは、社内の定義次第で数字が変わります。
安全弁がちゃんと効いているか確かめる
書いたつもりで効いていない防御ほど怖いものはありません。実際、この記事を書いている途中でタイムアウトが機能していないことに気づきました。
3段はそれぞれ守っているものが違うので、確かめ方も分ける必要があります。ツール経由のキーワード検査はFastMCPのin-memoryクライアントで、読み取り専用とタイムアウトはDB接続を直接触って確認します。
verify.py（検証スクリプトの全文コード（GitHub））
# 1段目: 接続そのものが書き換えを拒むか
conn = connect_readonly()
try:
conn.execute("DELETE FROM customers")
raise AssertionError("読み取り専用のはずが書き込めてしまいました")
except sqlite3.OperationalError as e:
assert "readonly" in str(e).lower()
# 3段目: 重いクエリが実時間で中断されるか
conn = connect_readonly()
install_query_timeout(conn, QUERY_TIMEOUT_SEC)
started = time.monotonic()
try:
conn.execute(HEAVY_SQL).fetchone()
raise AssertionError("中断されませんでした")
except sqlite3.OperationalError as e:
assert "interrupted" in str(e).lower()
print(f"{time.monotonic() - started:.1f} 秒で中断")筆者環境での結果です。
SELECTで始まってはいるがDROPを潜ませたクエリも止まっています。読み取り専用の接続は「attempt to write a readonly database」で書き込みを拒み、重いクエリは実時間10.0秒で中断されました。行数上限も、SELECT * FROM sales_transactionsを投げると50行のあとに「... 他 3950 件」と付いて返ってきます。
なおこの検証はin-memoryクライアントとDB接続を直接使ったもので、stdioやHTTPのトランスポートを経由した動作までは確認していません。
ここまで確認してから、はじめて社内で使ってもらうようにしてください。
HTTPサーバーとして常駐させる
ここまではstdio、つまりAIアプリが手元でサーバーを起動する形でした。独立して常駐させ、ネットワーク経由で使うならStreamable HTTPに切り替えます。変えるのは起動の1行だけです。
mcp_server_sales.py（起動部分のコード（GitHub））
mcp.run(transport="http", host="127.0.0.1", port=9904)これでhttp://127.0.0.1:9904/mcpがMCPエンドポイントになります。
待受は127.0.0.1にしてください。 このサーバーは認証を持っていません。0.0.0.0で待ち受けると、そのネットワークから届く相手なら誰でもSQLを実行できてしまいます。サンプルコードでも既定を127.0.0.1にして、外部待受を指定したときは警告を出すようにしました。社内で共有する、外から使うという段階に進むなら、HTTPS、認証と認可、接続元の制限、Origin検証、監査ログを前段に置く前提になります。この記事のサーバーは、その手前の「仕組みを理解するための最小構成」です。
死活監視用のエンドポイントも追加しておくと運用が楽です。
mcp_server_sales.py（ヘルスチェックのコード（GitHub））
@mcp.custom_route("/health", methods=["GET"])
async def health_check(request):
from starlette.responses import JSONResponse
return JSONResponse({"status": "ok", "service": "sales-database-server"})サンプルコードでは--httpオプションで切り替えられるようにしてあります。
python mcp_server_sales.py --http --port 9904起動したら、ヘルスチェックを叩いて生きていることを確かめます。
curl http://127.0.0.1:9904/health{"status":"ok","service":"sales-database-server"}返すのは状態だけにしてあります。監視に要らない情報（DBの絶対パスや例外の中身）を外へ出す理由はありません。
作ってみて分かったこと
今回いちばん効いたのは、ツールを2本に絞ったことでした。業務ごとに関数を並べるより、汎用のSQL実行ツールを1本渡すほうが、AIは自分で考えて答えを出してくれます。ただしこれは小規模なダミーDBでの話です。実データを扱うなら、取得できるテーブルや列、集計の粒度を絞った業務別ツールのほうが安全で、監査もしやすくなります。
そのぶん、descriptionの書き方で結果が大きく変わります。スキーマとクエリ例を丸ごと書いたら、AIのSQLが安定しました。descriptionは機能の説明だけでなく、ツールを使う条件や手順をモデルへ伝える場所としても働きます。ただし手順の指示はお願いであって、実行が保証される仕組みではありません。
安全側の手当ては3段にしました。読み取り専用で開き、SELECT以外を弾き、実行時間に上限を付ける。そして書いて終わりにせず、3段それぞれを実際に試して確かめました。ここを省くと、効いていない防御を効いていると思い込んだまま社内に配ることになります。実際、最初に書いたタイムアウトは実時間を測っておらず、まったく機能していませんでした。
返す行数の上限も忘れないでください。AIは平気で全件取得を投げてきます。50行で切って件数だけ正直に伝えたところ、筆者環境では集計クエリへ切り替える動きが確認できました。
常駐させて使うならStreamable HTTPに切り替えて、ヘルスチェックも足しておくと運用が楽になります。ただし認証は入っていないので、そのまま社内LANや外部へ出さないでください。
もうひとつ。このサンプルは仕組みを理解するための最小構成です。複数人が同時に使う場面では、同期的なDB処理をスレッドへ逃がす、同時実行数を絞る、DB側のタイムアウトを設ける、といった手当てが別に要ります。
次回予告
後編でやることは、はっきりしています。
手元（localhost）で作ったこのMCPサーバーを、Web版のChatGPT・Claudeと連携させます。OAuth認証の壁も乗り越えます。
CLI版のClaude Codeを使うのは開発者です。でもブラウザで使うWeb版なら、コマンドラインを触らない人にも渡せます。営業企画のメンバーが、いつものChatGPTの画面から「先月いちばん売れたGPUは？」と聞ける。同じMCPサーバーのまま、使える人の範囲が一気に広がります。
ところが、ここが本当に難しい。Web版のコネクタにMCPサーバーをつなぐには、次の2つが要ります。
ひとつは、インターネットから到達できるHTTPSのURL。手元のPCで動いているサーバーには、外から誰も届きません。かといって社内DBに届くサーバーをそのまま外に晒すわけにもいきません。
もうひとつが、利用者を識別してアクセスを制御するための認証・認可です。
ChatGPTもClaudeも、構成によっては認証なしのリモートMCPに接続できます。ただ、社内データベースに到達するMCPサーバーを認証なしで公開する選択肢はありません。そこで後編では、MCP Authorization仕様に沿ったOAuth認証を使います。
これを自前で実装するとなると、認可メタデータ、クライアント登録、認可エンドポイント、トークンの発行と検証と、ツール本体とはまったく別の実装が必要になります。ツール本体は200行で書けたのに、本題に入る前に力尽きる。ここで止まる人が多いところです。
後編では、この2つの壁をどう越えるかを、実際にWeb版のChatGPTとClaudeから今回の売上DBにつなぐところまで、画面つきで書きます。しかも無料で試せる方法でです。
先に開示しておくと、後編では当社が提供しているWireCanalを使います。無料で試せる枠があるので、そのまま手を動かせます（利用条件や無料枠は執筆時点のものです）。
それでは、また次回、お会いしましょう！
サンプルコード
この記事のコードはすべてGitHubに置いてあります。
qualiteg/mcp-server-tutorial ｜ サンプルコード一式とREADME（GitHub）
git clone https://github.com/qualiteg/mcp-server-tutorial.git
cd mcp-server-tutorial
python -m pip install -r requirements.txt
python db_setup.py続けてClaude Codeへ登録します。
claude mcp add sales-db -- python /path/to/mcp-server-tutorial/mcp_server_sales.py
claude mcp list
claude -p "職業別の売上トップ3を教えて"stdioで使う場合、python mcp_server_sales.py を別ターミナルで手動起動する必要はありません。登録したコマンドはClaude Codeが子プロセスとして起動します。
ここに書いた手順は、リポジトリのREADMEにも全部書いてあります。 クイックスタート、Claude Codeへの登録方法、提供しているツールの説明、HTTPサーバーとしての起動まで揃っているので、手元で動かすときはREADMEだけ見れば足ります。
