はじめに
こんばんは、mirukyです。
今回は、MCP（Model Context Protocol）サーバーをPythonで自作して、Claude Codeから自然言語で社内データベースを検索・分析できる環境を構築してみました。
「売上トップの社員を教えて」と入力するだけで、Claude CodeがSQLを自動生成→実行→結果を要約してくれます。MCPサーバーの自作は難しそうに聞こえますが、実際にやってみると Pythonコード約90行 で完成しました。本記事では、構築手順から実行結果までをまとめています。
MCPとは？ ― AIの「USB-C」ってよく言われますよね
MCP（Model Context Protocol） は、AIアプリと外部ツールを繋ぐオープン標準プロトコルです。
従来: AIアプリ ←→ 個別API実装 ←→ 各ツール（N×M問題）
MCP: AIアプリ ←→ MCP ←→ 各ツール（N+M で解決）
（よく聞く例えですが）USB-Cが充電ケーブルを統一したように、MCPはAIと外部システム間のインターフェースを統一します。
| 概念 | 役割 | REST APIとの対応 |
|---|---|---|
| Tools | LLMが呼び出せる関数 | POSTエンドポイント |
| Resources | 読み取り専用データ | GETエンドポイント |
| Prompts | 再利用可能なテンプレート | ― |
全体構成
今回は SQLite をモックの社内DBとして使い、MCPサーバー経由でClaude Codeから操作します。
環境構築
前提条件
- Python 3.10以上
- Claude Code がインストール済み
プロジェクト作成
# uvでプロジェクト初期化
uv init mcp-db-server
cd mcp-db-server
# MCP SDKをインストール
uv add "mcp[cli]"
# サーバーファイルを作成
touch server.py
touch init_db.py
uv はRust製の高速Pythonパッケージマネージャーです。未導入の場合は curl -LsSf https://astral.sh/uv/install.sh | sh でインストールできます。
Step 1: モック社内DBの作成
まず、社内DBを模したSQLiteデータベースを用意します。
init_db.py
import sqlite3
def init_database():
conn = sqlite3.connect("company.db")
cursor = conn.cursor()
# 社員テーブル
cursor.execute("""
CREATE TABLE IF NOT EXISTS employees (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
department TEXT NOT NULL,
position TEXT NOT NULL,
salary INTEGER NOT NULL,
hire_date TEXT NOT NULL
)
""")
# 売上テーブル
cursor.execute("""
CREATE TABLE IF NOT EXISTS sales (
id INTEGER PRIMARY KEY,
product_name TEXT NOT NULL,
amount INTEGER NOT NULL,
quantity INTEGER NOT NULL,
sale_date TEXT NOT NULL,
employee_id INTEGER,
FOREIGN KEY (employee_id) REFERENCES employees(id)
)
""")
# サンプルデータ投入
employees = [
(1, "田中太郎", "営業部", "部長", 8500000, "2015-04-01"),
(2, "佐藤花子", "営業部", "主任", 5500000, "2018-04-01"),
(3, "鈴木一郎", "開発部", "マネージャー", 7800000, "2016-07-01"),
(4, "高橋美咲", "開発部", "エンジニア", 6200000, "2020-04-01"),
(5, "伊藤健太", "人事部", "課長", 6800000, "2017-10-01"),
(6, "渡辺結衣", "営業部", "一般", 4200000, "2022-04-01"),
(7, "山本大輔", "開発部", "エンジニア", 5800000, "2021-04-01"),
(8, "中村さくら", "経理部", "主任", 5200000, "2019-04-01"),
]
sales = [
(1, "クラウドプランA", 1200000, 3, "2025-01-15", 1),
(2, "クラウドプランB", 800000, 5, "2025-01-20", 2),
(3, "オンプレライセンス", 3500000, 1, "2025-02-01", 1),
(4, "コンサルティング", 2000000, 2, "2025-02-10", 6),
(5, "クラウドプランA", 1200000, 4, "2025-03-05", 2),
(6, "保守サポート", 600000, 10, "2025-03-15", 1),
(7, "クラウドプランB", 800000, 7, "2025-04-01", 6),
(8, "開発受託", 5000000, 1, "2025-04-20", 3),
(9, "クラウドプランA", 1200000, 2, "2025-05-10", 2),
(10, "セキュリティ監査", 1500000, 3, "2025-05-25", 5),
]
cursor.executemany(
"INSERT OR REPLACE INTO employees VALUES (?, ?, ?, ?, ?, ?)", employees
)
cursor.executemany(
"INSERT OR REPLACE INTO sales VALUES (?, ?, ?, ?, ?, ?)", sales
)
conn.commit()
conn.close()
print("company.db を作成しました")
if __name__ == "__main__":
init_database()
uv run init_db.py
出力結果：
company.db を作成しました
Step 2: MCPサーバーの実装
ここが本記事の核心です。Python SDK の FastMCP を使って、わずか約90行でDBアクセス用MCPサーバーを構築します。
server.py
import sqlite3
import sys
from mcp.server.fastmcp import FastMCP
# ⚠️ STDIOサーバーでは stdout に書き込むとJSON-RPCが壊れる
# ログは必ず stderr に出力する
print("MCPサーバーを起動中...", file=sys.stderr)
# MCPサーバーの初期化
mcp = FastMCP("社内DB")
DB_PATH = "company.db"
def get_connection() -> sqlite3.Connection:
"""DB接続を取得"""
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
return conn
@mcp.tool()
def list_tables() -> str:
"""データベース内のテーブル一覧を取得する"""
conn = get_connection()
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
)
tables = [row["name"] for row in cursor.fetchall()]
conn.close()
return f"テーブル一覧: {', '.join(tables)}"
@mcp.tool()
def describe_table(table_name: str) -> str:
"""指定テーブルのカラム情報を取得する
Args:
table_name: 調べたいテーブル名
"""
conn = get_connection()
cursor = conn.execute(f"PRAGMA table_info({table_name})")
columns = cursor.fetchall()
conn.close()
if not columns:
return f"テーブル '{table_name}' が見つかりません"
result = f"テーブル '{table_name}' のスキーマ:\n"
for col in columns:
pk = " (PRIMARY KEY)" if col["pk"] else ""
nullable = "" if col["notnull"] else " NULL可"
result += f" - {col['name']}: {col['type']}{pk}{nullable}\n"
return result
@mcp.tool()
def query_db(sql: str) -> str:
"""SQLクエリを実行して結果を返す（SELECT文のみ）
Args:
sql: 実行するSELECT文
"""
# SELECT文のみ許可（安全対策）
if not sql.strip().upper().startswith("SELECT"):
return "エラー: SELECT文のみ実行可能です"
conn = get_connection()
try:
cursor = conn.execute(sql)
rows = cursor.fetchall()
if not rows:
return "結果: 0件"
# カラム名を取得
columns = [description[0] for description in cursor.description]
# 結果をテーブル形式で整形
result = f"結果: {len(rows)}件\n\n"
result += " | ".join(columns) + "\n"
result += "-" * (len(" | ".join(columns))) + "\n"
for row in rows:
result += " | ".join(str(row[col]) for col in columns) + "\n"
return result
except Exception as e:
return f"SQLエラー: {e}"
finally:
conn.close()
if __name__ == "__main__":
mcp.run(transport="stdio")
ポイント解説
| ポイント | 説明 |
|---|---|
@mcp.tool() |
関数をMCPツールとして公開。型ヒントとdocstringからツール定義が自動生成される |
transport="stdio" |
標準入出力でJSON-RPC通信。Claude Codeのローカル接続に最適 |
SELECT文のみ許可 |
安全対策としてデータ変更を防止 |
stderr へのログ出力 |
STDIOサーバーではstdoutに書くとプロトコルが壊れるため必須 |
Step 3: Claude Codeに接続
# MCPサーバーを登録
claude mcp add --transport stdio company-db -- uv run --directory /path/to/mcp-db-server server.py
--directory には 絶対パス を指定してください。pwd コマンドで確認できます。
登録確認:
claude mcp list
company-db: stdio ["/Users/you/.local/bin/uv", "run", "--directory", "/path/to/mcp-db-server", "server.py"]
Claude Codeを起動して /mcp コマンドでも接続状態を確認できます:
> /mcp
MCP Servers:
company-db: connected
Tools:
- list_tables
- describe_table
- query_db
3つのツールが認識されていれば準備完了です。
Step 4: 実際に使ってみる
例1: テーブル構造の把握
行った入力
> このDBにどんなテーブルがある？
出力結果：
⏺ データベースのテーブル構造を確認します。
[list_tables を実行]
テーブル一覧: employees, sales
[describe_table(table_name="employees") を実行]
テーブル 'employees' のスキーマ:
- id: INTEGER (PRIMARY KEY)
- name: TEXT
- department: TEXT
- position: TEXT
- salary: INTEGER
- hire_date: TEXT
[describe_table(table_name="sales") を実行]
テーブル 'sales' のスキーマ:
- id: INTEGER (PRIMARY KEY)
- product_name: TEXT
- amount: INTEGER
- quantity: INTEGER
- sale_date: TEXT
- employee_id: INTEGER (外部キー → employees.id)
このデータベースには2つのテーブルがあります：
■ employees - 社員情報（ID, 名前, 部署, 役職, 年収, 入社日）
■ sales - 売上データ（商品名, 金額, 数量, 日付, 担当社員ID）
例2: 自然言語で分析クエリ
行った入力
> 部署ごとの平均年収を高い順に出して
出力結果：
⏺ SQLクエリを生成して実行します。
[query_db を実行]
SQL: SELECT department, ROUND(AVG(salary)) as avg_salary
FROM employees GROUP BY department ORDER BY avg_salary DESC
結果: 4件
department | avg_salary
-----------+-----------
人事部 | 6800000
開発部 | 6600000
営業部 | 6066667
経理部 | 5200000
部署別の平均年収は上記の通りです。人事部が最も高く680万円、
経理部が最も低く520万円となっています。
例3: 複合的な分析
行った入力
> 2025年の月別売上推移と、最も売上に貢献した社員を教えて
出力結果
⏺ 2つのクエリを実行して分析します。
[query_db を実行（1/2）]
SQL: SELECT strftime('%Y-%m', sale_date) as month,
SUM(amount * quantity) as total_sales
FROM sales GROUP BY month ORDER BY month
結果: 5件
month | total_sales
--------+-----------
2025-01 | 7600000
2025-02 | 7500000
2025-03 | 10800000
2025-04 | 10600000
2025-05 | 6900000
[query_db を実行（2/2）]
SQL: SELECT e.name, SUM(s.amount * s.quantity) as total
FROM sales s JOIN employees e ON s.employee_id = e.id
GROUP BY e.name ORDER BY total DESC LIMIT 3
結果: 3件
name | total
---------+-----------
田中太郎 | 13100000
佐藤花子 | 11200000
渡辺結衣 | 9600000
【月別売上推移】
3月（1,080万円）と4月（1,060万円）が売上のピークです。
【売上貢献トップ3】
1位: 田中太郎 — ¥13,100,000
2位: 佐藤花子 — ¥11,200,000
3位: 渡辺結衣 — ¥9,600,000
田中太郎さん（営業部 部長）が最も売上に貢献しています。
自然言語だけで、JOIN付きの複合クエリが自動生成・実行される のがすごいですね。
セキュリティ上の注意点
本番環境で運用する場合、以下は 必須 です。
| 対策 | 理由 |
|---|---|
| SELECT文のみに制限 |
DROP TABLE や DELETE の実行を防止 |
| 読み取り専用ユーザーで接続 | DB側でも書き込み権限を遮断 |
| 接続先をレプリカDBに | 本番DBへの負荷を回避 |
| クエリのタイムアウト設定 | 重いクエリによるDB負荷を防止 |
| 機密カラムのマスク | 個人情報・パスワード等の露出を防止 |
| プロジェクトスコープで設定 |
.mcp.json でチーム共有＆バージョン管理 |
チーム共有する場合
プロジェクトルートに .mcp.json を配置すれば、チーム全員が同じMCPサーバー設定を利用できます:
.mcp.json
{
"mcpServers": {
"company-db": {
"command": "uv",
"args": ["run", "--directory", "/absolute/path/to/mcp-db-server", "server.py"],
"env": {
"DB_PATH": "path/to/readonly-replica.db"
}
}
}
}
もっとセキュリティに気をつけたい方は、こちらの記事もご覧ください。
【2026年最新版】Claude Codeで行うべきセキュリティ設定 10選
既存のDB MCP サーバーとの比較
「自作しなくても既存のものがあるのでは？」という疑問は当然です。
| 選択肢 | 特徴 | 自作との比較 |
|---|---|---|
| @bytebase/dbhub | PostgreSQL/MySQL/SQL Server対応の汎用サーバー | 汎用的だが社内ルール組み込み不可 |
| MCP Toolbox for Databases (Google) | BigQuery等Google Cloud系DB向け | GCP前提 |
| 自作 | 完全カスタマイズ可能 | クエリ制限・マスク等のセキュリティを柔軟に実装可能 |
自作のメリットは、社内独自のセキュリティポリシーやビジネスロジックを組み込める点 にあります。
例えば「特定テーブルへのアクセス制限」「機密カラムの自動マスク」「クエリログの監査」など、既存サーバーでは対応しきれない要件に向いています。
逆に、シンプルな用途なら既存サーバーで十分です。
# dbhubを使えば1行で済む
claude mcp add --transport stdio db -- npx -y @bytebase/dbhub --dsn "postgresql://readonly:pass@db.example.com:5432/analytics"
おわりに
ここまでお読みいただきありがとうございます。
今回は、MCPサーバーをPythonで自作し、Claude Codeから自然言語でDBを検索・分析できる環境を構築しました。
やってみて分かったのは、MCPサーバーの自作は思ったより簡単だということです。
@mcp.tool() デコレータで関数を公開するだけで、Claude Codeが勝手にツールを認識して適切に使ってくれます。
「非エンジニアの上司が、自然言語でDBを検索できる」― そんな未来が、わずか90行のPythonコードで実現できる時代になりました。
ではまた、お会いしましょう。