こんにちは！
生成AIや検索技術の進化が注目される中、Pythonで MCPサーバ を作成し、KIRO と連携して検索を可能にしました。本記事ではその仕組みとポイントを解説します。無料で行えるので、MCPサーバとKIROについて理解を深めたい方はぜひ行ってみてください。
MCPサーバーとは？
**MCP（Model Context Protocol）**は、AIアシスタントが外部のツールやデータソースと連携するための標準プロトコルです。
この記事では、社内の従業員情報や製品情報を検索できるMCPサーバーをPythonで作成します。
開発環境の準備
必要なもの
- Python 3.10以上
- テキストエディタ（VS Code推奨）
- uv（Pythonパッケージマネージャー）
uvのインストール
# pipでインストール
pip install uv
上記うまくいかない場合、下記でpythonを使って、Pythonモジュールとしてuvを実行インストール
python pip install uv
Step 1：プロジェクトの作成
# プロジェクトディレクトリを作成
mkdir company-info-mcp
cd company-info-mcp
# 仮想環境を作成
uv venv
# 仮想環境をアクティベート（コマンドプロンプト）
.venv\Scripts\activate.bat
# FastMCPをインストール
uv pip install fastmcp
Step 2：サンプルデータベースの作成
ファイル名: setup_database.py
import sqlite3
def create_sample_database():
"""サンプルの社内データベースを作成"""
conn = sqlite3.connect('company_database.db')
cursor = conn.cursor()
# 社員テーブルを作成
cursor.execute('''
CREATE TABLE IF NOT EXISTS employees (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
department TEXT NOT NULL,
email TEXT NOT NULL,
phone TEXT,
hire_date TEXT,
position TEXT
)
''')
# 製品テーブルを作成
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
category TEXT NOT NULL,
price INTEGER,
description TEXT,
stock INTEGER
)
''')
# サンプル社員データを挿入
employees = [
(1, '山田太郎', '営業部', 'yamada@company.com', '03-1234-5678', '2020-04-01', '営業課長'),
(2, '佐藤花子', '技術部', 'sato@company.com', '03-1234-5679', '2019-07-15', 'エンジニア'),
(3, '鈴木一郎', '営業部', 'suzuki@company.com', '03-1234-5680', '2021-01-10', '営業担当'),
(4, '田中美咲', '人事部', 'tanaka@company.com', '03-1234-5681', '2018-03-20', '人事マネージャー'),
(5, '高橋健太', '技術部', 'takahashi@company.com', '03-1234-5682', '2022-06-01', 'エンジニア'),
(6, '伊藤由美', 'マーケティング部', 'ito@company.com', '03-1234-5683', '2020-09-15', 'マーケター'),
(7, '渡辺誠', '総務部', 'watanabe@company.com', '03-1234-5684', '2017-11-01', '総務課長'),
(8, '中村愛', '技術部', 'nakamura@company.com', '03-1234-5685', '2021-04-01', 'シニアエンジニア'),
]
cursor.executemany('''
INSERT OR REPLACE INTO employees
(id, name, department, email, phone, hire_date, position)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', employees)
# サンプル製品データを挿入
products = [
(1, 'ノートPC Pro', 'ハードウェア', 150000, '高性能ビジネスノートPC', 50),
(2, 'オフィスチェア DX', '家具', 45000, 'エルゴノミクスデザインチェア', 30),
(3, 'プロジェクター 4K', '電子機器', 80000, '4K対応プロジェクター', 15),
(4, 'ワイヤレスマウス', '周辺機器', 3500, '静音設計マウス', 200),
(5, 'モニター 27インチ', 'ハードウェア', 35000, 'フルHDモニター', 80),
(6, 'キーボード メカニカル', '周辺機器', 12000, 'メカニカルキーボード', 100),
(7, 'Webカメラ HD', '電子機器', 8000, 'HD画質Webカメラ', 60),
(8, 'デスクライト LED', '家具', 6500, 'LED調光デスクライト', 120),
]
cursor.executemany('''
INSERT OR REPLACE INTO products
(id, name, category, price, description, stock)
VALUES (?, ?, ?, ?, ?, ?)
''', products)
conn.commit()
conn.close()
print("サンプルデータベースを作成しました: company_database.db")
if __name__ == "__main__":
create_sample_database()
今回はサンプル会社の社員データと、製品情報のデータを作成し使用します。
データベースを作成：
python setup_database.py
Step 3：MCPサーバーの作成
ファイル名: server.py
from fastmcp import FastMCP
import sqlite3
from typing import List, Dict, Optional
import requests
# MCPサーバーのインスタンスを作成
mcp = FastMCP("Company Info MCP Server")
DB_PATH = "company_database.db"
# ========== 社員情報検索ツール ==========
@mcp.tool()
def search_employees(name: str = None, department: str = None) -> List[Dict]:
"""
社員情報を検索します
Args:
name: 社員名（部分一致）
department: 部署名（完全一致）
Returns:
社員情報のリスト
"""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
query = "SELECT id, name, department, email, phone, position FROM employees WHERE 1=1"
params = []
if name:
query += " AND name LIKE ?"
params.append(f"%{name}%")
if department:
query += " AND department = ?"
params.append(department)
cursor.execute(query, params)
results = []
for row in cursor.fetchall():
results.append({
'id': row[0],
'name': row[1],
'department': row[2],
'email': row[3],
'phone': row[4],
'position': row[5]
})
conn.close()
return results
@mcp.tool()
def get_employee_details(employee_id: int) -> Dict:
"""
特定の社員の詳細情報を取得します
Args:
employee_id: 社員ID
Returns:
社員の詳細情報
"""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("""
SELECT id, name, department, email, phone, hire_date, position
FROM employees
WHERE id = ?
""", (employee_id,))
row = cursor.fetchone()
conn.close()
if not row:
return {"error": "社員が見つかりません"}
return {
'id': row[0],
'name': row[1],
'department': row[2],
'email': row[3],
'phone': row[4],
'hire_date': row[5],
'position': row[6]
}
@mcp.tool()
def list_departments() -> List[str]:
"""
部署の一覧を取得します
Returns:
部署名のリスト
"""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("SELECT DISTINCT department FROM employees ORDER BY department")
departments = [row[0] for row in cursor.fetchall()]
conn.close()
return departments
# ========== 製品情報検索ツール ==========
@mcp.tool()
def search_products(name: str = None, category: str = None, max_price: int = None) -> List[Dict]:
"""
製品情報を検索します
Args:
name: 製品名（部分一致）
category: カテゴリ名（完全一致）
max_price: 最大価格
Returns:
製品情報のリスト
"""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
query = "SELECT id, name, category, price, description, stock FROM products WHERE 1=1"
params = []
if name:
query += " AND name LIKE ?"
params.append(f"%{name}%")
if category:
query += " AND category = ?"
params.append(category)
if max_price:
query += " AND price <= ?"
params.append(max_price)
cursor.execute(query, params)
results = []
for row in cursor.fetchall():
results.append({
'id': row[0],
'name': row[1],
'category': row[2],
'price': row[3],
'description': row[4],
'stock': row[5]
})
conn.close()
return results
@mcp.tool()
def get_product_details(product_id: int) -> Dict:
"""
特定の製品の詳細情報を取得します
Args:
product_id: 製品ID
Returns:
製品の詳細情報
"""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("""
SELECT id, name, category, price, description, stock
FROM products
WHERE id = ?
""", (product_id,))
row = cursor.fetchone()
conn.close()
if not row:
return {"error": "製品が見つかりません"}
return {
'id': row[0],
'name': row[1],
'category': row[2],
'price': row[3],
'description': row[4],
'stock': row[5]
}
@mcp.tool()
def list_categories() -> List[str]:
"""
製品カテゴリの一覧を取得します
Returns:
カテゴリ名のリスト
"""
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("SELECT DISTINCT category FROM products ORDER BY category")
categories = [row[0] for row in cursor.fetchall()]
conn.close()
return categories
# ========== 外部API連携サンプル ==========
@mcp.tool()
def get_weather_info(city: str) -> Dict:
"""
外部APIから天気情報を取得します（サンプル実装）
実際に使用する場合は、OpenWeatherMap等のAPIキーを取得して設定してください
Args:
city: 都市名
Returns:
天気情報
"""
# サンプル実装（実際のAPI呼び出しはコメントアウト）
# API_KEY = "your_api_key_here"
# url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={API_KEY}&lang=ja"
# response = requests.get(url)
# return response.json()
# デモ用のダミーデータを返す
return {
"city": city,
"weather": "晴れ",
"temperature": "25°C",
"humidity": "60%",
"note": "これはサンプルデータです。実際のAPIを使用する場合は、APIキーを設定してください。"
}
@mcp.tool()
def call_external_api_sample(endpoint: str, params: Dict = None) -> Dict:
"""
外部APIを呼び出すサンプル
実際に使用する場合は、認証情報や適切なエラーハンドリングを追加してください
Args:
endpoint: APIエンドポイントURL
params: リクエストパラメータ
Returns:
APIレスポンス
"""
try:
# サンプル実装
# response = requests.get(endpoint, params=params, timeout=10)
# response.raise_for_status()
# return response.json()
return {
"status": "success",
"message": "これはサンプル実装です",
"endpoint": endpoint,
"params": params,
"note": "実際のAPI呼び出しを行う場合は、requests.get()のコメントを解除してください"
}
except Exception as e:
return {
"status": "error",
"message": str(e)
}
# サーバーを起動
if __name__ == "__main__":
mcp.run()
今回はすべて無料で行いたいため、API呼び出しは実装していません。
実際にAPIを使ったMCPサーバを作成したい場合はコメントにも記載ある通り、認証情報や適切なエラーハンドリングを追加してください。
Step 4：サーバーをテスト
# サーバーを起動
python server.py
正常に起動すれば、MCPサーバーが待機状態になります。
Step 5：Kiroに接続する
mcp.json の設定
ファイルパス: .kiro/settings/mcp.json
{
"mcpServers": {
"company-info": {
"command": "python",
"args": ["C:/path/to/company-info-mcp/server.py"],
"disabled": false,
"autoApprove": []
}
}
}
パスの設定方法：
-
C:/path/to/company-info-mcp/を実際のプロジェクトパスに変更 - 例:
C:/Users/YourName/Documents/company-info-mcp/server.py
接続手順
-
.kiro/settings/mcp.jsonを作成・編集 - Kiroを再起動、またはコマンドパレット → "MCP: Reconnect Servers"
- 左側のMCPパネルで「company-info」が表示されることを確認
また、MCPサーバの接続を終了したい場合には
MCP: Disconnect All Servers
KIROでコマンドパレット(Ctrl+Shift+P)を開き、上記コマンドを入力することで、全てのMCPサーバの接続が切断されます。
使用例
Kiroのチャットで以下のように質問できます：
「営業部の社員を検索して」
「山田さんの詳細情報を教えて」
「部署の一覧を表示して」
「ノートPCを検索して」
「3万円以下の製品を探して」
「製品カテゴリを教えて」
営業部の社員を検索してと入力すると、KIROがMCPサーバに接続して、回答してくれるようになっています。
よくある問題
問題1: データベースが見つからない
解決方法: setup_database.pyを実行してデータベースを作成
問題2: サーバーが起動しない
# Pythonのパスを確認
where python
# パッケージを再インストール
uv pip install --force-reinstall fastmcp
問題3: Kiroでツールが表示されない
解決方法:
1. mcp.jsonのパスが正しいか確認
2. Kiroのログを確認（出力パネル → Kiro）
3. サーバーを直接実行してエラーを確認
おわりに
お疲れ様でした！これで社内情報をKiroで検索できるようになりましたね。
最初は「本当に動くのかな？」と不安だったかもしれませんが、実際に動いてみるとその便利さに驚くはずです。サンプルデータから始めて、少しずつ実際の業務データに置き換えていってください。
※実際のデータに置き換える場合にはセキュリティには十分ご注意ください
またpythonで実装できることならなんでも実行可能なので、データベースの更新、API連携の実装等も可能ですので試してみてください。
MCPサーバーを作成することで、Kiroに独自の機能を追加することができます。実際の社内業務システムと連携し、様々なデータをKIROから自然言語で検索できるようになれば日々の業務効率が大きく上昇するはずです。
ここまで読んでくださりありがとうございました！
何か困ったことがあれば、コメントで質問してください！