次の方法で共有


App Service アプリを GitHub Copilot チャット用 MCP サーバーとして統合する (Python)

このチュートリアルでは、モデル コンテキスト プロトコル (MCP) を使用して FastAPI アプリの機能を公開し、GitHub Copilot にツールとして追加し、Copilot チャット エージェント モードで自然言語を使用してアプリと対話する方法について説明します。

GitHub Copilot チャット ウィンドウの MCP ツール呼び出しからの応答を示すスクリーンショット。

Web アプリケーションにショッピング、ホテル予約、データ管理などの便利な機能が既にある場合は、これらの機能を次の目的で簡単に利用できます。

MCP サーバーを Web アプリに追加することで、エージェントがユーザー プロンプトに応答したときにアプリの機能を理解して使用できるようになります。 つまり、アプリでできることは何でも、エージェントでも実行できます。

  • MCP サーバーを Web アプリに追加します。
  • GitHub Copilot Chat エージェント モードで MCP サーバーをローカルでテストします。
  • MCP サーバーを Azure App Service にデプロイし、GitHub Copilot Chat でそれに接続します。

[前提条件]

このチュートリアルでは、「Azure で PostgreSQL を使用して Python FastAPI Web アプリをデプロイする」で使用するサンプルを使用していることを前提としています。

少なくとも、GitHub Codespaces で サンプル アプリケーション を開き、 azd upを実行してアプリをデプロイします。

GitHub codespaces で開く で開く

MCP サーバーを Web アプリに追加する

  1. 次の例に示すように、codespace エクスプローラーで src/pyproject.toml を開き、依存関係の一覧に mcp[cli] を追加します。

    dependencies = [
        ...
        "mcp[cli]",
    ]
    
  2. src/fastapi_app で、mcp_server.pyという名前ファイルを作成し、次の MCP サーバー初期化コードをファイルに貼り付けます。

    import asyncio
    import contextlib
    from contextlib import asynccontextmanager
    
    from mcp.server.fastmcp import FastMCP
    from sqlalchemy.sql import func
    from sqlmodel import Session, select
    
    from .models import Restaurant, Review, engine
    
    # Create a FastMCP server. Use stateless_http=True for simple mounting. Default path is .../mcp
    mcp = FastMCP("RestaurantReviewsMCP", stateless_http=True)
    
    # Lifespan context manager to start/stop the MCP session manager with the FastAPI app
    @asynccontextmanager
    async def mcp_lifespan(app):
        async with contextlib.AsyncExitStack() as stack:
            await stack.enter_async_context(mcp.session_manager.run())
            yield
    
    # MCP tool: List all restaurants with their average rating and review count
    @mcp.tool()
    async def list_restaurants_mcp() -> list[dict]:
        """List restaurants with their average rating and review count."""
    
        def sync():
            with Session(engine) as session:
                statement = (
                    select(
                        Restaurant,
                        func.avg(Review.rating).label("avg_rating"),
                        func.count(Review.id).label("review_count"),
                    )
                    .outerjoin(Review, Review.restaurant == Restaurant.id)
                    .group_by(Restaurant.id)
                )
                results = session.exec(statement).all()
                rows = []
                for restaurant, avg_rating, review_count in results:
                    r = restaurant.dict()
                    r["avg_rating"] = float(avg_rating) if avg_rating is not None else None
                    r["review_count"] = review_count
                    r["stars_percent"] = (
                        round((float(avg_rating) / 5.0) * 100) if review_count > 0 and avg_rating is not None else 0
                    )
                    rows.append(r)
                return rows
    
        return await asyncio.to_thread(sync)
    
    # MCP tool: Get a restaurant and all its reviews by restaurant_id
    @mcp.tool()
    async def get_details_mcp(restaurant_id: int) -> dict:
        """Return the restaurant and its related reviews as objects."""
    
        def sync():
            with Session(engine) as session:
                restaurant = session.exec(select(Restaurant).where(Restaurant.id == restaurant_id)).first()
                if restaurant is None:
                    return None
                reviews = session.exec(select(Review).where(Review.restaurant == restaurant_id)).all()
                return {"restaurant": restaurant.dict(), "reviews": [r.dict() for r in reviews]}
    
        return await asyncio.to_thread(sync)
    
    # MCP tool: Create a new review for a restaurant
    @mcp.tool()
    async def create_review_mcp(restaurant_id: int, user_name: str, rating: int, review_text: str) -> dict:
        """Create a new review for a restaurant and return the created review dict."""
    
        def sync():
            with Session(engine) as session:
                review = Review()
                review.restaurant = restaurant_id
                review.review_date = __import__("datetime").datetime.now()
                review.user_name = user_name
                review.rating = int(rating)
                review.review_text = review_text
                session.add(review)
                session.commit()
                session.refresh(review)
                return review.dict()
    
        return await asyncio.to_thread(sync)
    
    # MCP tool: Create a new restaurant
    @mcp.tool()
    async def create_restaurant_mcp(restaurant_name: str, street_address: str, description: str) -> dict:
        """Create a new restaurant and return the created restaurant dict."""
    
        def sync():
            with Session(engine) as session:
                restaurant = Restaurant()
                restaurant.name = restaurant_name
                restaurant.street_address = street_address
                restaurant.description = description
                session.add(restaurant)
                session.commit()
                session.refresh(restaurant)
                return restaurant.dict()
    
        return await asyncio.to_thread(sync)
    

    FastMCP() 初期化子は、MCP Python SDK のステートレス モード パターンを使用して MCP サーバーを作成します。 既定では、ストリーミング可能な HTTP エンドポイントは /mcp サブパスに設定されます。

    • @mcp.tool() デコレーターは、その実装を使用して MCP サーバーにツールを追加します。
    • ツール関数の説明は、呼び出し元エージェントがツールとそのパラメーターの使用方法を理解するのに役立ちます。

    ツールは、フォーム ベースの FastAPI Web アプリケーションの既存のレストラン レビュー機能を複製します。 必要に応じて、更新および削除機能用のツールをさらに追加できます。

  3. src/fastapi_app/app.py で、app = FastAPI()の行 (24 行目) を見つけて、次のコードに置き換えます。

    from .mcp_server import mcp, mcp_lifespan
    app = FastAPI(lifespan=mcp_lifespan)
    app.mount("/mcp", mcp.streamable_http_app())
    

    このコードは、MCP サーバーのストリーミング可能な HTTP エンドポイントを、 /mcpパスにある既存の FastAPI アプリにマウントします。 ストリーミング可能な HTTP エンドポイントの既定のパスと共に、完全なパスは /mcp/mcp

MCP サーバーをローカルでテストする

  1. codespace ターミナルで、次のコマンドを使用してアプリケーションを実行します。

    python3 -m venv .venv
    source .venv/bin/activate
    pip install -r src/requirements.txt
    pip install -e src
    python3 src/fastapi_app/seed_data.py
    python3 -m uvicorn fastapi_app:app --reload --port=8000
    
  2. [ ブラウザーで開く] を選択し、いくつかのレストランとレビューを追加します。

    uvicornは実行したままにします。 MCP サーバーは現在、 http://localhost:8000/mcp/mcp で実行されています。

  3. コードスペースに戻り、Copilot Chat を開き、プロンプト ボックスで [エージェント モード] を選択します。

  4. [ツール] ボタンを選択し、ポップアップの右上隅にある [MCP サーバーの追加] アイコンを選択します。

    GitHub Copilot チャット エージェント モードで MCP サーバーを追加する方法を示すスクリーンショット。

  5. HTTP (HTTP または Server-Sent イベント) を選択します。

  6. [ サーバー URL の入力] に「 http://localhost:8000/mcp/mcp」と入力します。

  7. [ サーバー ID の入力] に、restaurant_ratings または任意の名前を入力します。

  8. [ ワークスペースの設定] を選択します

  9. 新しい Copilot チャット ウィンドウで、「Show me the restaurant ratings」のように入力します。

  10. 既定では、MCP サーバーを呼び出すと、GitHub Copilot にセキュリティ確認が表示されます。 続行を選択します。

    GitHub Copilot Chat での MCP 呼び出しからの既定のセキュリティ メッセージを示すスクリーンショット。

    MCP ツールの呼び出しが成功したことを示す応答が表示されます。

    GitHub Copilot チャット ウィンドウの MCP ツール呼び出しからの応答を示すスクリーンショット。

MCP サーバーを App Service にデプロイする

  1. codespace ターミナルに戻り、変更をコミットして変更をデプロイするか (GitHub Actions メソッド)、または azd up (Azure Developer CLI メソッド) を実行します。

  2. AZD 出力で、アプリの URL を見つけます。 URL は、AZD の出力では次のようになります。

     Deploying services (azd deploy)
    
       (✓) Done: Deploying service web
       - Endpoint: <app-url>
     
  3. azd upが完了したら、.vscode/mcp.jsonを開きます。 URL を <app-url>/mcp/mcpに変更します。

  4. 変更した MCP サーバー構成の上で、[開始] を選択 します

    ローカル mcp.json ファイルから MCP サーバーを手動で起動する方法を示すスクリーンショット。

  5. 新しい GitHub Copilot チャット ウィンドウを開始します。 レストランの評価を表示したり、Copilot エージェントで新しいレストランや新しい評価を作成したりできます。

セキュリティのベスト プラクティス

MCP サーバーが大規模言語モデル (LLM) を使用するエージェントによって呼び出される場合は、 迅速なインジェクション 攻撃に注意してください。 次のセキュリティのベスト プラクティスを検討してください。

  • 認証と承認: 承認されたユーザーまたはエージェントのみがツールにアクセスできるように、Microsoft Entra 認証を使用して MCP サーバーをセキュリティで保護します。 詳細なガイドについては、 Microsoft Entra 認証を使用した Visual Studio Code からの Azure App Service へのセキュリティで保護されたモデル コンテキスト プロトコルの呼び出し に関するページを参照してください。
  • 入力の検証とサニタイズ: 無効または悪意のある入力を防ぐために、常に受信データを検証します。 Python アプリの場合は、 Pydantic などのライブラリを使用して、専用の入力モデル (RestaurantCreate や ReviewCreate など) でデータ検証規則を適用します。 ベスト プラクティスと実装の詳細については、ドキュメントを参照してください。
  • HTTPS: このサンプルは、既定で HTTPS を適用し、転送中のデータを暗号化するための無料の TLS/SSL 証明書を提供する Azure App Service に依存しています。
  • 最小限の特権の原則: ユース ケースに必要なツールとデータのみを公開します。 必要な場合を除き、機密性の高い操作を公開しないでください。
  • レート制限と調整: API Management またはカスタム ミドルウェアを使用して、不正使用やサービス拒否攻撃を防ぎます。
  • ログ記録と監視: 監査と異常検出のための MCP エンドポイントのログ アクセスと使用状況。 疑わしいアクティビティを監視します。
  • CORS 構成: MCP サーバーがブラウザーからアクセスされる場合は、クロスオリジン要求を信頼されたドメインに制限します。 詳細については、「 CORS を有効にする」を参照してください。
  • 定期的な更新: 既知の脆弱性を軽減するために、依存関係を最新の状態に保ちます。

その他のリソース

AI を Azure App Service アプリケーションに統合する