共用方式為


將 App Service 應用程式整合為適用於 GitHub Copilot Chat 的 MCP 伺服器 (Python)

在本教學課程中,可瞭解如何透過模型內容通訊協定 (MCP) 公開 FastAPI 應用程式的功能、做為工具新增至 GitHub Copilot,以及在 Copilot Chat 代理模式中以自然語言與應用程式互動。

螢幕擷取畫面顯示 GitHub Copilot Chat 視窗中的 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. 返回 codespace,開啟 Copilot Chat,然後在提示方塊中選取 [代理程式] 模式。

  4. 選取 [工具] 按鈕,然後選取快顯視窗右上角的 [新增 MCP 伺服器] 圖示。

    螢幕擷取畫面,顯示如何在 GitHub Copilot Chat 代理模式中新增 MCP 伺服器。

  5. 選取 [HTTP (HTTP 或伺服器傳送的事件)]

  6. 在 [輸入伺服器 URL] 輸入 http://localhost:8000/mcp/mcp

  7. 在 [輸入伺服器識別碼] 輸入 restaurant_ratings 或其他任何名稱。

  8. 選取 [工作區設定]

  9. 在新的 Copilot Chat 視窗中,輸入「告訴我這家餐廳的評分」之類的內容。

  10. 根據預設,GitHub Copilot 會在您叫用 MCP 伺服器時顯示安全性確認通知。 選取繼續

    螢幕擷取畫面,顯示 GitHub Copilot Chat 中 MCP 叫用的預設安全性訊息。

    現在應該會看到回應,顯示 MCP 工具呼叫成功。

    螢幕擷取畫面顯示 GitHub Copilot Chat 視窗中的 MCP 工具呼叫回應。

將 MCP 伺服器部署至 App Service

  1. 返回 codespace 終端機,藉由認可變更 (GitHub Actions 方法) 或執行 azd up (Azure 開發人員 CLI 方法) 來部署變更。

  2. 在 AZD 輸出結果中,尋找應用程式的 URL。 AZD 輸出中 URL 看起來像這樣:

     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 Chat 視窗。 這時應該能夠查看餐廳評分,也可以在 Copilot 代理程式中建立新的餐廳和評分。

安全性最佳做法

如果呼叫 MCP 伺服器的是大型語言模型 (LLM) 支援的代理程式,請留意提示插入類型的攻擊。 建議採取以下安全性最佳做法:

  • 認證與授權:以 Microsoft Entra 認證保護你的 MCP 伺服器,確保只有授權使用者或代理能存取你的工具。 請參閱透過 Microsoft Entra 驗證,從 Visual Studio Code 向 Azure App Service 發出安全的模型內容通訊協定呼叫 (機器翻譯)
  • 輸入驗證和清理:一律驗證傳入的資料,以免輸入無效或惡意輸入。 對於 Python 應用程式,請使用 Pydantic 等程式庫,透過專用輸入模型 (例如 RestaurantCreate 和 ReviewCreate) 強制執行資料驗證規則。 請參閱其文件,以取得最佳做法和實作詳細資料。
  • HTTPS:此範例依賴 Azure App Service,預設會強制執行 HTTPS,並提供免費的 TLS/SSL 憑證來加密傳輸中的資料。
  • 最小權限原則:僅公開使用案例必需的工具和資料。 除非必要,否則請不要公開敏感性操作。
  • 速率限制和節流:使用 API 管理或自訂的中介軟體來防堵濫用情形和阻斷服務攻擊。
  • 事件記錄和監控:記錄 MCP 端點的存取和使用情況,以供稽核和異常偵測。 請監控可疑活動。
  • CORS 設定:如果 MCP 伺服器是從瀏覽器存取,請將跨來源請求範圍限制在受信任的網域。 如需詳細資訊,請參閱啟用 CORS
  • 定期更新:相依性需保持最新狀態,以減輕已知漏洞的影響。

更多資源

將 AI 整合至 Azure App Service 應用程式