0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

MCPを実装理解!Python + FastAPIでAIアシスタントを開発するチュートリアル

0
Posted at

はじめに

最近、技術記事サイトのランキングが MCP という技術の記事ばかりで埋まっています。

「MCP?こんなに話題なら勉強しておかないといけない」

解説記事を読んでもなんとなくしかわからない。やはり実装して処理レベルで理解する方が早いと思ったので、チュートリアルとしてまとめました。

この記事では MCP を活用しながら AIと対話しながら操作できる TODO アプリを開発します。

技術スタックは以下です:

  • パッケージ管理 : uv
  • API サーバー : Python + FastAPI + SQLite
  • MCP サーバー : Python + MCP SDK
  • 動作確認 : MCP Inspector(ブラウザベース)

対象者

  • MCP がいまいちわからない人
  • 実装して深く理解したい人
  • Python で Web API を作ってみたい人
  • AI アプリを開発したい人

MCP とは?

MCP とは Model Context Protocol の略称です。

簡単に言うと、LLM(大規模言語モデル)と外部ツールを繋ぐための標準プロトコルです。

スクリーンショット 2026-07-03 224653.png

例えばチャットで「TODOに掃除を追加して」と AI に指示したとき、以下のことが起きます:

  1. AI がどのツールを使うべきか判断する
  2. MCP 経由で TODO 追加ツールを呼び出す
  3. API サーバーにリクエストが飛び DB にレコードが追加される

MCP がある世界では、AI アプリ側はどのサービスを使うか・どんな API 仕様かを意識することなく、一貫した呼び出し方だけ覚えれば OK になります。


今回実装するアプリの仕組み

スクリーンショット 2026-07-03 224553.png


MCP の通信方式

スクリーンショット 2026-07-03 224701.png

MCP クライアントとサーバーの通信方式は主に 3 つあります。

方式 特徴
Stdio 標準入出力で通信。ローカル向け
Streamable HTTP 単一の HTTP エンドポイントでストリーミング。現在推奨
HTTP + SSE SSE でリアルタイム配信

今回は HTTP + SSE で実装します。


環境準備

uv のインストール

uv は Python の高速パッケージマネージャーです。pip より圧倒的に速く、仮想環境の管理も一括でできます。

curl -LsSf https://astral.sh/uv/install.sh | sh

インストール後、パスを通します:

source $HOME/.local/bin/env
# または
export PATH="$HOME/.local/bin:$PATH"

バージョン確認:

uv --version

Python のインストール(uv 経由)

uv python install 3.12

1. API サーバーの構築

まずは FastAPI で TODO の CRUD API を作ります。MCP サーバーはこの API を叩いて TODO を操作します。

プロジェクト作成

mkdir mcp-todos
cd mcp-todos

# API プロジェクト初期化
uv init api
cd api

依存関係の追加

uv add fastapi uvicorn httpx

api/main.py を作成

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import sqlite3
from typing import Optional

app = FastAPI()

# CORS 設定(MCP Inspector からのアクセスを許可)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

DB_PATH = "todos.db"

# DB 初期化
def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS todos (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            completed INTEGER DEFAULT 0
        )
    """)
    conn.commit()
    conn.close()

init_db()

def get_conn():
    return sqlite3.connect(DB_PATH)

# リクエストモデル
class TodoCreate(BaseModel):
    title: str

class TodoUpdate(BaseModel):
    title: Optional[str] = None
    completed: Optional[bool] = None

# TODO 一覧取得
@app.get("/todos")
def get_todos():
    conn = get_conn()
    rows = conn.execute("SELECT id, title, completed FROM todos").fetchall()
    conn.close()
    return [{"id": r[0], "title": r[1], "completed": bool(r[2])} for r in rows]

# TODO 追加
@app.post("/todos")
def create_todo(body: TodoCreate):
    conn = get_conn()
    cur = conn.execute("INSERT INTO todos (title) VALUES (?)", (body.title,))
    conn.commit()
    todo_id = cur.lastrowid
    conn.close()
    return {"id": todo_id, "title": body.title, "completed": False}

# TODO 更新
@app.put("/todos/{todo_id}")
def update_todo(todo_id: int, body: TodoUpdate):
    conn = get_conn()
    row = conn.execute("SELECT id FROM todos WHERE id = ?", (todo_id,)).fetchone()
    if not row:
        raise HTTPException(status_code=404, detail="Todo not found")
    if body.title is not None:
        conn.execute("UPDATE todos SET title = ? WHERE id = ?", (body.title, todo_id))
    if body.completed is not None:
        conn.execute("UPDATE todos SET completed = ? WHERE id = ?", (int(body.completed), todo_id))
    conn.commit()
    row = conn.execute("SELECT id, title, completed FROM todos WHERE id = ?", (todo_id,)).fetchone()
    conn.close()
    return {"id": row[0], "title": row[1], "completed": bool(row[2])}

# TODO 削除
@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
    conn = get_conn()
    row = conn.execute("SELECT id FROM todos WHERE id = ?", (todo_id,)).fetchone()
    if not row:
        raise HTTPException(status_code=404, detail="Todo not found")
    conn.execute("DELETE FROM todos WHERE id = ?", (todo_id,))
    conn.commit()
    conn.close()
    return {"success": True}

API サーバーを起動

uv run uvicorn main:app --port 8080 --reload

動作確認

# TODO 追加
curl -X POST http://localhost:8080/todos \
  -H "Content-Type: application/json" \
  -d '{"title": "掃除をする"}'

# TODO 一覧取得
curl http://localhost:8080/todos

# TODO 完了に更新
curl -X PUT http://localhost:8080/todos/1 \
  -H "Content-Type: application/json" \
  -d '{"completed": true}'

# TODO 削除
curl -X DELETE http://localhost:8080/todos/1

2. MCP サーバーの構築

次に MCP サーバーを実装します。ここが今回のチュートリアルの本質です。

MCP サーバーにツールを登録すると、AI はそのツール一覧を見てどのツールを使うか自分で判断します。

プロジェクト作成

cd ../  # mcp-todos ディレクトリに戻る

# MCP プロジェクト初期化
uv init mcp-server
cd mcp-server

依存関係の追加

uv add mcp httpx

mcp-server/main.py を作成

import httpx
from mcp.server.fastmcp import FastMCP

API_BASE = "http://localhost:8080"

mcp = FastMCP("todo-mcp-server")

@mcp.tool(description="新しい TODO を追加する")
async def addTodoItem(title: str) -> str:
    async with httpx.AsyncClient() as client:
        await client.post(f"{API_BASE}/todos", json={"title": title})
    return f"{title} を追加しました"

@mcp.tool(description="指定した ID の TODO を削除する")
async def deleteTodoItem(id: int) -> str:
    async with httpx.AsyncClient() as client:
        await client.delete(f"{API_BASE}/todos/{id}")
    return f"ID {id} を削除しました"

@mcp.tool(description="指定した ID の TODO を更新する")
async def updateTodoItem(id: int, title: str = None, completed: bool = None) -> str:
    body = {}
    if title is not None:
        body["title"] = title
    if completed is not None:
        body["completed"] = completed
    async with httpx.AsyncClient() as client:
        await client.put(f"{API_BASE}/todos/{id}", json=body)
    return f"ID {id} を更新しました"

app = mcp.streamable_http_app()

MCP サーバーを起動

uv run uvicorn main:app --port 3001 --reload

3. テストクライアントで動作確認

Python の対話式クライアントを作って MCP サーバーの動作を確認します。
Termux だけで完結するので Android でもそのまま使えます。

プロジェクト作成

cd ../  # mcp-todos ディレクトリに戻る

uv init test-client
cd test-client

依存関係の追加

uv add httpx

依存関係の追加

uv add mcp

test-client/main.py を作成

import asyncio
from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession

MCP_URL = "http://localhost:3001/mcp"

async def main():
    print("=== MCP Todo クライアント ===\n")

    async with streamablehttp_client(MCP_URL) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            while True:
                print("操作を選んでください:")
                print("1. ツール一覧を見る")
                print("2. TODO を追加する")
                print("3. TODO を更新する")
                print("4. TODO を削除する")
                print("0. 終了")
                choice = input("\n番号を入力: ").strip()

                if choice == "0":
                    print("終了します")
                    break

                elif choice == "1":
                    result = await session.list_tools()
                    print("\n--- ツール一覧 ---")
                    for t in result.tools:
                        print(f"{t.name} : {t.description}")
                    print()

                elif choice == "2":
                    title = input("追加する TODO のタイトル: ").strip()
                    result = await session.call_tool("addTodoItem", {"title": title})
                    print(f"\n{result.content[0].text}\n")

                elif choice == "3":
                    todo_id = int(input("更新する TODO の ID: ").strip())
                    print("何を更新する?")
                    print("1. タイトル")
                    print("2. 完了状態")
                    print("3. 両方")
                    sub = input("番号: ").strip()
                    args = {"id": todo_id}
                    if sub in ("1", "3"):
                        args["title"] = input("新しいタイトル: ").strip()
                    if sub in ("2", "3"):
                        args["completed"] = input("完了にする? (y/n): ").strip().lower() == "y"
                    result = await session.call_tool("updateTodoItem", args)
                    print(f"\n{result.content[0].text}\n")

                elif choice == "4":
                    todo_id = int(input("削除する TODO の ID: ").strip())
                    result = await session.call_tool("deleteTodoItem", {"id": todo_id})
                    print(f"\n{result.content[0].text}\n")

                else:
                    print("無効な番号です\n")

if __name__ == "__main__":
    asyncio.run(main())

起動前の準備

テストクライアントを実行する前に、別のターミナルで API サーバーと MCP サーバーを起動しておきます。

ターミナル 1 : API サーバー

cd mcp-todos/api
uv run uvicorn main:app --port 8080 --reload

ターミナル 2 : MCP サーバー

cd mcp-todos/mcp-server
uv run uvicorn main:app --port 3001 --reload

ターミナル 3 : テストクライアント

cd mcp-todos/test-client
uv run main.py

動作テスト

起動すると以下のメニューが表示されます:

=== MCP Todo クライアント ===

操作を選んでください:
1. ツール一覧を見る
2. TODO を追加する
3. TODO を更新する
4. TODO を削除する
0. 終了

TODO を追加する:

番号を入力: 2
追加する TODO のタイトル: 掃除をする
✓ 掃除をする を追加しました

DB に反映されているか確認:

curl http://localhost:8080/todos

TODO を完了にする:

番号を入力: 3
更新する TODO の ID: 1
何を更新する?
1. タイトル
2. 完了状態
3. 両方
番号: 2
完了にする? (y/n): y
✓ ID 1 を更新しました

TODO を削除する:

番号を入力: 4
削除する TODO の ID: 1
✓ ID 1 を削除しました

MCP のポイントまとめ

要素 役割
ツール名 AI がツールを特定するための名前
description AI がどのツールを使うか判断するための説明
inputSchema ツールに渡す引数のバリデーション
call_tool 実際にツールが呼ばれたときの処理

description が重要で、ここの書き方次第で AI が正しくツールを選んでくれるかどうかが変わります。


まとめ

今回の構成は以下の通りでした:

テストクライアント(Python)
    ↕ SSE
MCP サーバー(Python・port 3001)
    ↕ HTTP
API サーバー(FastAPI・port 8080)
    ↕
SQLite

uv を使うことで仮想環境の管理やパッケージインストールがシンプルになります。また Python を使うことで Termux でも詰まらずに MCP の本質部分を学べます。

MCP を理解すると、あとはどんな外部ツールを繋ぐかだけの話になります。Google Drive、Slack、GitHub など、MCP サーバーさえ実装すれば AI が自動で使いこなしてくれます。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?