สร้าง Agent วางแผนการทดสอบ QA โดยใช้ ADK, MCP และ Gemini 2.5 Flash ที่มีโหมดการคิด

1. บทนำ

fd4ad67ab57d9257.png

ศักยภาพของการใช้ Generative AI ในการสร้างแผนการทดสอบมาจากความสามารถในการแก้ปัญหาที่ท้าทายที่สุด 2 อย่างในการประกันคุณภาพสมัยใหม่ ได้แก่ ความเร็วและความครอบคลุม ในวงจร Agile และ DevOps ที่รวดเร็วในปัจจุบัน การเขียนแผนการทดสอบแบบละเอียดด้วยตนเองถือเป็นอุปสรรคสำคัญที่ทำให้กระบวนการทดสอบทั้งหมดล่าช้า เอเจนต์ที่ทำงานด้วย Gen AI สามารถรับเรื่องราวของผู้ใช้และข้อกำหนดทางเทคนิคเพื่อสร้างแผนการทดสอบที่ครอบคลุมได้ภายในไม่กี่นาที ไม่กี่วัน เพื่อให้กระบวนการ QA เป็นไปตามการพัฒนา นอกจากนี้ AI ยังเก่งในการระบุสถานการณ์ที่ซับซ้อน กรณีข้อยกเว้น และเส้นทางเชิงลบที่มนุษย์อาจมองข้ามไป ซึ่งจะช่วยปรับปรุงความครอบคลุมของการทดสอบได้อย่างมาก และลดข้อบกพร่องที่หลุดรอดไปสู่การใช้งานจริงได้อย่างมีนัยสำคัญ

ในโค้ดแล็บนี้ เราจะมาดูวิธีสร้างเอเจนต์ดังกล่าวที่สามารถดึงเอกสารข้อกำหนดของผลิตภัณฑ์จาก Confluence และให้ความคิดเห็นที่เป็นประโยชน์ รวมถึงสร้างแผนการทดสอบที่ครอบคลุมซึ่งส่งออกเป็นไฟล์ CSV ได้

ใน Codelab นี้ คุณจะได้ใช้แนวทางแบบทีละขั้นตอนดังนี้

  1. เตรียมโปรเจ็กต์ Google Cloud และเปิดใช้ API ที่จำเป็นทั้งหมดในโปรเจ็กต์
  2. ตั้งค่าพื้นที่ทํางานสําหรับสภาพแวดล้อมการเขียนโค้ด
  3. การเตรียม mcp-server ในเครื่องสำหรับ Confluence
  4. การจัดโครงสร้างซอร์สโค้ด พรอมต์ และเครื่องมือของเอเจนต์ ADK เพื่อเชื่อมต่อกับเซิร์ฟเวอร์ MCP
  5. ทำความเข้าใจการใช้งานบริบทของบริการและเครื่องมือของอาร์ติแฟกต์
  6. การทดสอบเอเจนต์โดยใช้ UI การพัฒนาเว็บในเครื่องของ ADK
  7. จัดการตัวแปรสภาพแวดล้อมและตั้งค่าไฟล์ที่จำเป็นสำหรับการติดตั้งใช้งานแอปพลิเคชันใน Cloud Run
  8. ทำให้แอปพลิเคชันใช้งานได้ใน Cloud Run

ภาพรวมสถาปัตยกรรม

819179179d974c07.jpeg

ข้อกำหนดเบื้องต้น

  • คุ้นเคยกับการทำงานด้วย Python
  • ความเข้าใจเกี่ยวกับสถาปัตยกรรม Full-Stack พื้นฐานโดยใช้บริการ HTTP

สิ่งที่คุณจะได้เรียนรู้

  • การออกแบบสถาปัตยกรรมของตัวแทน ADK ขณะใช้ความสามารถต่างๆ
  • การใช้เครื่องมือกับเครื่องมือที่กำหนดเองและ MCP
  • การตั้งค่าเอาต์พุตไฟล์โดยตัวแทนโดยใช้การจัดการบริการ Artifact
  • การใช้ BuiltInPlanner เพื่อปรับปรุงการทำงานโดยการวางแผนด้วยความสามารถด้านการคิดของ Gemini 2.5 Flash
  • การโต้ตอบและการแก้ไขข้อบกพร่องผ่านอินเทอร์เฟซเว็บของ ADK
  • ทำให้แอปพลิเคชันใช้งานได้กับ Cloud Run โดยใช้ Dockerfile และระบุตัวแปรสภาพแวดล้อม

สิ่งที่ต้องมี

  • เว็บเบราว์เซอร์ Chrome
  • บัญชี Gmail
  • โปรเจ็กต์ Cloud ที่เปิดใช้การเรียกเก็บเงิน
  • (ไม่บังคับ) พื้นที่ Confluence ที่มีหน้าเอกสารเกี่ยวกับข้อกำหนดสำหรับผลิตภัณฑ์

Codelab นี้ออกแบบมาสำหรับนักพัฒนาซอฟต์แวร์ทุกระดับ (รวมถึงผู้เริ่มต้น) โดยใช้ Python ในแอปพลิเคชันตัวอย่าง อย่างไรก็ตาม คุณไม่จำเป็นต้องมีความรู้เกี่ยวกับ Python เพื่อทำความเข้าใจแนวคิดที่นำเสนอ ไม่ต้องกังวลเรื่องพื้นที่ Confluence หากคุณไม่มี เราจะให้ข้อมูลเข้าสู่ระบบเพื่อลองใช้ Codelab นี้

2. ก่อนเริ่มต้น

เลือกโปรเจ็กต์ที่ใช้งานอยู่ใน Cloud Console

Codelab นี้ถือว่าคุณมีโปรเจ็กต์ Google Cloud ที่เปิดใช้การเรียกเก็บเงินอยู่แล้ว หากยังไม่มี ให้ทำตามวิธีการด้านล่างเพื่อเริ่มต้นใช้งาน

  1. ในคอนโซล Google Cloud ให้เลือกหรือสร้างโปรเจ็กต์ Google Cloud ในหน้าตัวเลือกโปรเจ็กต์
  2. ตรวจสอบว่าได้เปิดใช้การเรียกเก็บเงินสำหรับโปรเจ็กต์ Cloud แล้ว ดูวิธีตรวจสอบว่าได้เปิดใช้การเรียกเก็บเงินในโปรเจ็กต์แล้วหรือไม่

3a29a3ec1306e9b4.png

ตั้งค่าโปรเจ็กต์ Cloud ในเทอร์มินัล Cloud Shell

  1. คุณจะใช้ Cloud Shell ซึ่งเป็นสภาพแวดล้อมบรรทัดคำสั่งที่ทำงานใน Google Cloud คลิกเปิดใช้งาน Cloud Shell ที่ด้านบนของคอนโซล Google Cloud

1829c3759227c19b.png

  1. เมื่อเชื่อมต่อกับ Cloud Shell แล้ว ให้ตรวจสอบว่าคุณได้รับการตรวจสอบสิทธิ์แล้วและตั้งค่าโปรเจ็กต์เป็นรหัสโปรเจ็กต์ของคุณโดยใช้คำสั่งต่อไปนี้
gcloud auth list
  1. เรียกใช้คำสั่งต่อไปนี้ใน Cloud Shell เพื่อยืนยันว่าคำสั่ง gcloud รู้จักโปรเจ็กต์ของคุณ
gcloud config list project
  1. หากไม่ได้ตั้งค่าโปรเจ็กต์ ให้ใช้คำสั่งต่อไปนี้เพื่อตั้งค่า
gcloud config set project <YOUR_PROJECT_ID>

หรือคุณจะดู PROJECT_ID id ในคอนโซลก็ได้

4032c45803813f30.jpeg

คลิก แล้วคุณจะเห็นโปรเจ็กต์ทั้งหมดและรหัสโปรเจ็กต์ทางด้านขวา

8dc17eb4271de6b5.jpeg

  1. เปิดใช้ API ที่จำเป็นผ่านคำสั่งที่แสดงด้านล่าง การดำเนินการนี้อาจใช้เวลาสักครู่ โปรดอดทนรอ
gcloud services enable aiplatform.googleapis.com \
                       run.googleapis.com \
                       cloudbuild.googleapis.com \
                       cloudresourcemanager.googleapis.com

เมื่อเรียกใช้คำสั่งสำเร็จ คุณควรเห็นข้อความที่คล้ายกับข้อความที่แสดงด้านล่าง

Operation "operations/..." finished successfully.

คุณสามารถใช้คอนโซลแทนคำสั่ง gcloud ได้โดยค้นหาแต่ละผลิตภัณฑ์หรือใช้ลิงก์นี้

หากพลาด API ใดไป คุณก็เปิดใช้ได้เสมอในระหว่างการติดตั้งใช้งาน

โปรดดูคำสั่งและการใช้งาน gcloud ในเอกสารประกอบ

ไปที่ Cloud Shell Editor และตั้งค่าไดเรกทอรีการทำงานของแอปพลิเคชัน

ตอนนี้เราสามารถตั้งค่าโปรแกรมแก้ไขโค้ดเพื่อเขียนโค้ดได้แล้ว เราจะใช้ Cloud Shell Editor สำหรับการดำเนินการนี้

  1. คลิกปุ่ม "เปิดเครื่องมือแก้ไข" ซึ่งจะเปิด Cloud Shell Editor และเราจะเขียนโค้ดที่นี่ได้ b16d56e4979ec951.png
  2. ตรวจสอบว่าได้ตั้งค่าโปรเจ็กต์ Cloud Code ไว้ที่มุมล่างซ้าย (แถบสถานะ) ของโปรแกรมแก้ไข Cloud Shell ตามที่ไฮไลต์ไว้ในรูปภาพด้านล่าง และตั้งค่าเป็นโปรเจ็กต์ Google Cloud ที่ใช้งานอยู่ซึ่งคุณเปิดใช้การเรียกเก็บเงินไว้ ให้สิทธิ์หากได้รับข้อความแจ้ง หากคุณทำตามคำสั่งก่อนหน้านี้แล้ว ปุ่มอาจชี้ไปยังโปรเจ็กต์ที่เปิดใช้งานโดยตรงแทนปุ่มลงชื่อเข้าใช้

f5003b9c38b43262.png

  1. จากนั้นโคลนไดเรกทอรีการทำงานของเทมเพลตสำหรับโค้ดแล็บนี้จาก Github โดยเรียกใช้คำสั่งต่อไปนี้ ซึ่งจะสร้างไดเรกทอรีการทำงานในไดเรกทอรี qa-test-planner-agent
git clone https://github.com/alphinside/qa-test-planner-agent.git qa-test-planner-agent
  1. หลังจากนั้น ให้ไปที่ส่วนบนสุดของ Cloud Shell Editor แล้วคลิกไฟล์->เปิดโฟลเดอร์ ค้นหาไดเรกทอรีชื่อผู้ใช้ แล้วค้นหาไดเรกทอรี qa-test-planner-agent จากนั้นคลิกปุ่มตกลง ซึ่งจะทำให้ไดเรกทอรีที่เลือกเป็นไดเรกทอรีการทำงานหลัก ในตัวอย่างนี้ ชื่อผู้ใช้คือ alvinprayuda ดังนั้นเส้นทางไดเรกทอรีจึงแสดงอยู่ด้านล่าง

2c53696f81d805cc.png

fb321426f3c848b3.png

ตอนนี้ Cloud Shell Editor ควรมีลักษณะดังนี้

9ae344bc522d2a6f.png

การตั้งค่าสภาพแวดล้อม

เตรียมสภาพแวดล้อมเสมือนของ Python

ขั้นตอนถัดไปคือการเตรียมสภาพแวดล้อมในการพัฒนา เทอร์มินัลที่ใช้งานอยู่ปัจจุบันควรอยู่ในไดเรกทอรีการทำงาน qa-test-planner-agent เราจะใช้ Python 3.12 ใน Codelab นี้ และจะใช้ตัวจัดการโปรเจ็กต์ Python ของ uv เพื่อลดความซับซ้อนในการสร้างและจัดการเวอร์ชัน Python และสภาพแวดล้อมเสมือน

  1. หากยังไม่ได้เปิดเทอร์มินัล ให้เปิดโดยคลิกเทอร์มินัล -> เทอร์มินัลใหม่ หรือใช้ Ctrl + Shift + C ซึ่งจะเปิดหน้าต่างเทอร์มินัลที่ส่วนล่างของเบราว์เซอร์

f8457daf0bed059e.jpeg

  1. ดาวน์โหลด uv และติดตั้ง Python 3.12 ด้วยคำสั่งต่อไปนี้
curl -LsSf https://astral.sh/uv/0.7.19/install.sh | sh && \
source $HOME/.local/bin/env && \
uv python install 3.12
  1. ตอนนี้มาเริ่มต้นสภาพแวดล้อมเสมือนโดยใช้ uv กัน เรียกใช้คำสั่งนี้
uv sync --frozen

การดำเนินการนี้จะสร้างไดเรกทอรี .venv และติดตั้งทรัพยากร Dependency การดูpyproject.toml อย่างรวดเร็วจะให้ข้อมูลเกี่ยวกับทรัพยากร Dependency ที่แสดงดังนี้

dependencies = [
    "google-adk>=1.5.0",
    "mcp-atlassian>=0.11.9",
    "pandas>=2.3.0",
    "python-dotenv>=1.1.1",
]
  1. หากต้องการทดสอบสภาพแวดล้อมเสมือน ให้สร้างไฟล์ใหม่ชื่อ main.py แล้วคัดลอกโค้ดต่อไปนี้
def main():
   print("Hello from qa-test-planner-agent")

if __name__ == "__main__":
   main()
  1. จากนั้นเรียกใช้คำสั่งต่อไปนี้
uv run main.py

คุณจะได้รับเอาต์พุตดังที่แสดงด้านล่าง

Using CPython 3.12
Creating virtual environment at: .venv
Hello from qa-test-planner-agent!

ซึ่งแสดงว่าระบบกำลังตั้งค่าโปรเจ็กต์ Python อย่างถูกต้อง

ตอนนี้เราจะไปที่ขั้นตอนถัดไปได้แล้ว นั่นคือการสร้างเอเจนต์และบริการ

3. สร้างเอเจนต์โดยใช้ Google ADK และ Gemini 2.5

ข้อมูลเบื้องต้นเกี่ยวกับโครงสร้างไดเรกทอรี ADK

มาเริ่มสำรวจสิ่งที่ ADK มีให้และวิธีสร้างเอเจนต์กัน คุณเข้าถึงเอกสารประกอบทั้งหมดของ ADK ได้ในURL นี้ ADK มีเครื่องมือมากมายให้เราใช้ในการดำเนินการคำสั่ง CLI ตัวอย่างเช่น

  • ตั้งค่าโครงสร้างไดเรกทอรีของเอเจนต์
  • ลองโต้ตอบอย่างรวดเร็วผ่านอินพุตเอาต์พุตของ CLI
  • ตั้งค่าอินเทอร์เฟซเว็บ UI การพัฒนาในเครื่องอย่างรวดเร็ว

ตอนนี้มาสร้างโครงสร้างไดเรกทอรีของเอเจนต์โดยใช้คำสั่ง CLI กัน เรียกใช้คำสั่งต่อไปนี้

uv run adk create qa_test_planner \
   --model gemini-2.5-flash \
   --project {your-project-id} \
   --region global

ซึ่งจะสร้างโครงสร้างไดเรกทอรีของ Agent ต่อไปนี้ในไดเรกทอรีที่ใช้งานอยู่ในปัจจุบัน

qa_test_planner/
├── __init__.py
├── .env
├── agent.py

และหากตรวจสอบ init.py และ agent.py คุณจะเห็นโค้ดนี้

# __init__.py

from . import agent

# agent.py

from google.adk.agents import Agent

root_agent = Agent(
    model='gemini-2.5-flash',
    name='root_agent',
    description='A helpful assistant for user questions.',
    instruction='Answer user questions to the best of your knowledge',
)

การสร้างเอเจนต์เครื่องมือวางแผนการทดสอบ QA

มาสร้างเอเจนต์วางแผนการทดสอบ QA กันเลย เปิดไฟล์ qa_test_planner/agent.py แล้วคัดลอกโค้ดด้านล่างซึ่งจะมี root_agent

# qa_test_planner/agent.py

from google.adk.agents import Agent
from google.adk.tools.mcp_tool.mcp_toolset import (
    MCPToolset,
    StdioConnectionParams,
    StdioServerParameters,
)
from google.adk.planners import BuiltInPlanner
from google.genai import types
from dotenv import load_dotenv
import os
from pathlib import Path
from pydantic import BaseModel
from typing import Literal
import tempfile
import pandas as pd
from google.adk.tools import ToolContext


load_dotenv(dotenv_path=Path(__file__).parent / ".env")

confluence_tool = MCPToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="uvx",
            args=[
                "mcp-atlassian",
                f"--confluence-url={os.getenv('CONFLUENCE_URL')}",
                f"--confluence-username={os.getenv('CONFLUENCE_USERNAME')}",
                f"--confluence-token={os.getenv('CONFLUENCE_TOKEN')}",
                "--enabled-tools=confluence_search,confluence_get_page,confluence_get_page_children",
            ],
            env={},
        ),
        timeout=60,
    ),
)


class TestPlan(BaseModel):
    test_case_key: str
    test_type: Literal["manual", "automatic"]
    summary: str
    preconditions: str
    test_steps: str
    expected_result: str
    associated_requirements: str


async def write_test_tool(
    prd_id: str, test_cases: list[dict], tool_context: ToolContext
):
    """A tool to write the test plan into file

    Args:
        prd_id: Product requirement document ID
        test_cases: List of test case dictionaries that should conform to these fields:
            - test_case_key: str
            - test_type: Literal["manual","automatic"]
            - summary: str
            - preconditions: str
            - test_steps: str
            - expected_result: str
            - associated_requirements: str

    Returns:
        A message indicating success or failure of the validation and writing process
    """
    validated_test_cases = []
    validation_errors = []

    # Validate each test case
    for i, test_case in enumerate(test_cases):
        try:
            validated_test_case = TestPlan(**test_case)
            validated_test_cases.append(validated_test_case)
        except Exception as e:
            validation_errors.append(f"Error in test case {i + 1}: {str(e)}")

    # If validation errors exist, return error message
    if validation_errors:
        return {
            "status": "error",
            "message": "Validation failed",
            "errors": validation_errors,
        }

    # Write validated test cases to CSV
    try:
        # Convert validated test cases to a pandas DataFrame
        data = []
        for tc in validated_test_cases:
            data.append(
                {
                    "Test Case ID": tc.test_case_key,
                    "Type": tc.test_type,
                    "Summary": tc.summary,
                    "Preconditions": tc.preconditions,
                    "Test Steps": tc.test_steps,
                    "Expected Result": tc.expected_result,
                    "Associated Requirements": tc.associated_requirements,
                }
            )

        # Create DataFrame from the test case data
        df = pd.DataFrame(data)

        if not df.empty:
            # Create a temporary file with .csv extension
            with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as temp_file:
                # Write DataFrame to the temporary CSV file
                df.to_csv(temp_file.name, index=False)
                temp_file_path = temp_file.name

            # Read the file bytes from the temporary file
            with open(temp_file_path, "rb") as f:
                file_bytes = f.read()

            # Create an artifact with the file bytes
            await tool_context.save_artifact(
                filename=f"{prd_id}_test_plan.csv",
                artifact=types.Part.from_bytes(data=file_bytes, mime_type="text/csv"),
            )

            # Clean up the temporary file
            os.unlink(temp_file_path)

            return {
                "status": "success",
                "message": (
                    f"Successfully wrote {len(validated_test_cases)} test cases to "
                    f"CSV file: {prd_id}_test_plan.csv"
                ),
            }
        else:
            return {"status": "warning", "message": "No test cases to write"}
    except Exception as e:
        return {
            "status": "error",
            "message": f"An error occurred while writing to CSV: {str(e)}",
        }


root_agent = Agent(
    model="gemini-2.5-flash",
    name="qa_test_planner_agent",
    description="You are an expert QA Test Planner and Product Manager assistant",
    instruction=f"""
Help user search any product requirement documents on Confluence. Furthermore you also can provide the following capabilities when asked:
- evaluate product requirement documents and assess it, then give expert input on what can be improved 
- create a comprehensive test plan following Jira Xray mandatory field formatting, result showed as markdown table. Each test plan must also have explicit mapping on 
    which user stories or requirements identifier it's associated to 

Here is the Confluence space ID with it's respective document grouping:

- "{os.getenv("CONFLUENCE_PRD_SPACE_ID")}" : space to store Product Requirements Documents

Do not making things up, Always stick to the fact based on data you retrieve via tools.
""",
    tools=[confluence_tool, write_test_tool],
    planner=BuiltInPlanner(
        thinking_config=types.ThinkingConfig(
            include_thoughts=True,
            thinking_budget=2048,
        )
    ),
)

ตั้งค่าไฟล์การกำหนดค่า

ตอนนี้เราจะต้องเพิ่มการตั้งค่าเพิ่มเติมสำหรับโปรเจ็กต์นี้ เนื่องจากเอเจนต์นี้จะต้องมีสิทธิ์เข้าถึง Confluence

เปิด qa_test_planner/.env และค่าตัวแปรสภาพแวดล้อมต่อไปนี้ แล้วตรวจสอบว่าไฟล์ .env ที่ได้มีลักษณะดังนี้

GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT={YOUR-CLOUD-PROJECT-ID}
GOOGLE_CLOUD_LOCATION=global
CONFLUENCE_URL={YOUR-CONFLUENCE-DOMAIN}
CONFLUENCE_USERNAME={YOUR-CONFLUENCE-USERNAME}
CONFLUENCE_TOKEN={YOUR-CONFLUENCE-API-TOKEN}
CONFLUENCE_PRD_SPACE_ID={YOUR-CONFLUENCE-SPACE-ID}

อย่างไรก็ตาม เราไม่สามารถทำให้พื้นที่ Confluence นี้เป็นแบบสาธารณะได้ ดังนั้นคุณจึงตรวจสอบไฟล์เหล่านี้เพื่ออ่านเอกสารข้อกำหนดของผลิตภัณฑ์ที่มีอยู่ได้โดยใช้ข้อมูลเข้าสู่ระบบข้างต้น

คำอธิบายโค้ด

สคริปต์นี้มีการเริ่มต้นตัวแทนที่เราเริ่มต้นสิ่งต่อไปนี้

  • ตั้งค่าโมเดลที่จะใช้เป็น gemini-2.5-flash
  • ตั้งค่าเครื่องมือ MCP ของ Confluence ซึ่งจะสื่อสารผ่าน Stdio
  • ตั้งค่าเครื่องมือที่กำหนดเอง write_test_tool เพื่อเขียนแผนการทดสอบและส่งออก CSV ไปยังอาร์ติแฟกต์
  • ตั้งค่าคำอธิบายและวิธีการของเอเจนต์
  • เปิดใช้การวางแผนก่อนสร้างคำตอบสุดท้ายหรือการดำเนินการโดยใช้ความสามารถในการคิดของ Gemini 2.5 Flash

เมื่อโมเดล Gemini ขับเคลื่อนเอเจนต์ด้วยความสามารถในการคิดในตัวและกำหนดค่าด้วยอาร์กิวเมนต์เครื่องมือวางแผน เอเจนต์จะแสดงความสามารถในการคิดและแสดงในอินเทอร์เฟซเว็บได้ด้วย โค้ดที่ใช้กำหนดค่านี้แสดงอยู่ด้านล่าง

# qa-test-planner/agent.py

from google.adk.planners import BuiltInPlanner
from google.genai import types

...

# Provide the confluence tool to agent

root_agent = Agent(
    model="gemini-2.5-flash",
    name="qa_test_planner_agent",
    ...,
    tools=[confluence_tool, write_test_tool],
    planner=BuiltInPlanner(
        thinking_config=types.ThinkingConfig(
            include_thoughts=True,
            thinking_budget=2048,
        )
    ),

...

และก่อนที่จะดำเนินการ เราจะเห็นกระบวนการคิดของโมเดล

185371e0e7e5995e.png

เครื่องมือ Confluence MCP

หากต้องการเชื่อมต่อกับเซิร์ฟเวอร์ MCP จาก ADK เราต้องใช้ MCPToolSet ซึ่งนำเข้าจากโมดูล google.adk.tools.mcp_tool.mcp_toolset ได้ โค้ดในการเริ่มต้นที่แสดงด้านล่าง ( ตัดทอนเพื่อประสิทธิภาพ)

# qa-test-planner/agent.py

from google.adk.tools.mcp_tool.mcp_toolset import (
    MCPToolset,
    StdioConnectionParams,
    StdioServerParameters,
)

...

# Initialize the Confluence MCP Tool via Stdio Output

confluence_tool = MCPToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="uvx",
            args=[
                "mcp-atlassian",
                f"--confluence-url={os.getenv('CONFLUENCE_URL')}",
                f"--confluence-username={os.getenv('CONFLUENCE_USERNAME')}",
                f"--confluence-token={os.getenv('CONFLUENCE_TOKEN')}",
                "--enabled-tools=confluence_search,confluence_get_page,confluence_get_page_children",
            ],
            env={},
        ),
        timeout=60,
    ),
)


...

# Provide the confluence tool to agent

root_agent = Agent(
    model="gemini-2.5-flash",
    name="qa_test_planner_agent",
    ...,
    tools=[confluence_tool, write_test_tool],

...

การกำหนดค่านี้จะทำให้เอเจนต์เริ่มต้น Confluence MCP Server เป็นกระบวนการแยกต่างหาก และจะจัดการการสื่อสารกับกระบวนการเหล่านั้นผ่าน Studio I/O ภาพสถาปัตยกรรม MCP ต่อไปนี้แสดงโฟลว์นี้ โดยมีเครื่องหมายอยู่ในกรอบสีแดงด้านล่าง

85d51458405f9217.png

นอกจากนี้ ภายในอาร์กิวเมนต์คำสั่งของการเริ่มต้น MCP เรายังจำกัดเครื่องมือที่ใช้ได้ให้เป็นเครื่องมือต่อไปนี้เท่านั้น ได้แก่ confluence_search, confluence_get_page และ confluence_get_page_children ซึ่งรองรับกรณีการใช้งานของเอเจนต์ทดสอบ QA เราใช้เซิร์ฟเวอร์ Atlassian MCP ที่ชุมชนมีส่วนร่วม ( ดูรายละเอียดเพิ่มเติมในเอกสารประกอบฉบับเต็ม) สำหรับบทแนะนำ Codelab นี้

เครื่องมือเขียนทดสอบ

หลังจากที่เอเจนต์ได้รับบริบทจากเครื่องมือ Confluence MCP แล้ว เอเจนต์จะสร้างแผนการทดสอบที่จำเป็นสำหรับผู้ใช้ได้ แต่เราต้องการสร้างไฟล์ที่มีแผนการทดสอบนี้เพื่อให้สามารถบันทึกและแชร์กับบุคคลอื่นได้ เราจึงมีเครื่องมือที่กำหนดเอง write_test_tool ด้านล่างเพื่อรองรับการดำเนินการนี้

# qa-test-planner/agent.py

...

async def write_test_tool(
    prd_id: str, test_cases: list[dict], tool_context: ToolContext
):
    """A tool to write the test plan into file

    Args:
        prd_id: Product requirement document ID
        test_cases: List of test case dictionaries that should conform to these fields:
            - test_case_key: str
            - test_type: Literal["manual","automatic"]
            - summary: str
            - preconditions: str
            - test_steps: str
            - expected_result: str
            - associated_requirements: str

    Returns:
        A message indicating success or failure of the validation and writing process
    """
    validated_test_cases = []
    validation_errors = []

    # Validate each test case
    for i, test_case in enumerate(test_cases):
        try:
            validated_test_case = TestPlan(**test_case)
            validated_test_cases.append(validated_test_case)
        except Exception as e:
            validation_errors.append(f"Error in test case {i + 1}: {str(e)}")

    # If validation errors exist, return error message
    if validation_errors:
        return {
            "status": "error",
            "message": "Validation failed",
            "errors": validation_errors,
        }

    # Write validated test cases to CSV
    try:
        # Convert validated test cases to a pandas DataFrame
        data = []
        for tc in validated_test_cases:
            data.append(
                {
                    "Test Case ID": tc.test_case_key,
                    "Type": tc.test_type,
                    "Summary": tc.summary,
                    "Preconditions": tc.preconditions,
                    "Test Steps": tc.test_steps,
                    "Expected Result": tc.expected_result,
                    "Associated Requirements": tc.associated_requirements,
                }
            )

        # Create DataFrame from the test case data
        df = pd.DataFrame(data)

        if not df.empty:
            # Create a temporary file with .csv extension
            with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as temp_file:
                # Write DataFrame to the temporary CSV file
                df.to_csv(temp_file.name, index=False)
                temp_file_path = temp_file.name

            # Read the file bytes from the temporary file
            with open(temp_file_path, "rb") as f:
                file_bytes = f.read()

            # Create an artifact with the file bytes
            await tool_context.save_artifact(
                filename=f"{prd_id}_test_plan.csv",
                artifact=types.Part.from_bytes(data=file_bytes, mime_type="text/csv"),
            )

            # Clean up the temporary file
            os.unlink(temp_file_path)

            return {
                "status": "success",
                "message": (
                    f"Successfully wrote {len(validated_test_cases)} test cases to "
                    f"CSV file: {prd_id}_test_plan.csv"
                ),
            }
        else:
            return {"status": "warning", "message": "No test cases to write"}
    except Exception as e:
        return {
            "status": "error",
            "message": f"An error occurred while writing to CSV: {str(e)}",
        }


...

ฟังก์ชันที่ประกาศไว้ข้างต้นมีไว้เพื่อรองรับฟังก์ชันการทำงานต่อไปนี้

  1. ตรวจสอบแผนการทดสอบที่สร้างขึ้นเพื่อให้เป็นไปตามข้อกำหนดของฟิลด์ที่บังคับ เราตรวจสอบโดยใช้โมเดล Pydantic และหากเกิดข้อผิดพลาด เราจะส่งข้อความแสดงข้อผิดพลาดกลับไปยังตัวแทน
  2. ส่งออกผลลัพธ์เป็น CSV โดยใช้ฟังก์ชันการทำงานของ Pandas
  3. จากนั้นระบบจะบันทึกไฟล์ที่สร้างขึ้นเป็นอาร์ติแฟกต์โดยใช้ความสามารถของ Artifact Service ซึ่งเข้าถึงได้โดยใช้ออบเจ็กต์ ToolContext ที่เข้าถึงได้ในการเรียกใช้เครื่องมือทุกครั้ง

หากเราบันทึกไฟล์ที่สร้างขึ้นเป็นอาร์ติแฟกต์ ระบบจะทําเครื่องหมายเป็นเหตุการณ์ในรันไทม์ ADK และจะแสดงในการโต้ตอบของเอเจนต์ในภายหลังในอินเทอร์เฟซเว็บ

57f5ab95f4ae86e7.png

ด้วยวิธีนี้ เราจึงตั้งค่าการตอบกลับไฟล์จากตัวแทนแบบไดนามิกเพื่อให้แก่ผู้ใช้ได้

4. การทดสอบตัวแทน

ตอนนี้มาลองสื่อสารกับเอเจนต์ผ่าน CLI กัน โดยเรียกใช้คำสั่งต่อไปนี้

uv run adk run qa_test_planner

โดยจะแสดงเอาต์พุตในลักษณะนี้ ซึ่งคุณจะแชทกับตัวแทนได้ แต่จะส่งข้อความได้เฉพาะผ่านอินเทอร์เฟซนี้เท่านั้น

Log setup complete: /tmp/agents_log/agent.xxxx_xxx.log
To access latest log: tail -F /tmp/agents_log/agent.latest.log
Running agent qa_test_planner_agent, type exit to exit.
user: hello
[qa_test_planner_agent]: Hello there! How can I help you today?
user:

เราดีใจที่ได้แชทกับตัวแทนผ่าน CLI แต่จะดียิ่งขึ้นไปอีกหากเรามีแชทบนเว็บที่ใช้งานง่าย ซึ่งเราก็ทำได้เช่นกัน นอกจากนี้ ADK ยังช่วยให้เรามี UI สำหรับการพัฒนาเพื่อโต้ตอบและตรวจสอบสิ่งที่เกิดขึ้นระหว่างการโต้ตอบ เรียกใช้คำสั่งต่อไปนี้เพื่อเริ่มเซิร์ฟเวอร์ UI สำหรับการพัฒนาในเครื่อง

uv run adk web --port 8080

ซึ่งจะสร้างเอาต์พุตคล้ายกับตัวอย่างต่อไปนี้ ซึ่งหมายความว่าเราเข้าถึงอินเทอร์เฟซเว็บได้แล้ว

INFO:     Started server process [xxxx]
INFO:     Waiting for application startup.

+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://localhost:8080.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)

ตอนนี้หากต้องการตรวจสอบ ให้คลิกปุ่มตัวอย่างเว็บที่ด้านบนของ Cloud Shell Editor แล้วเลือกแสดงตัวอย่างบนพอร์ต 8080

e7c9f56c2463164.png

คุณจะเห็นหน้าเว็บต่อไปนี้ ซึ่งคุณสามารถเลือกตัวแทนที่พร้อมให้บริการได้ที่ปุ่มเมนูแบบเลื่อนลงด้านซ้ายบน ( ในกรณีของเราควรเป็น qa_test_planner) และโต้ตอบกับบ็อต คุณจะเห็นข้อมูลมากมายเกี่ยวกับรายละเอียดบันทึกในระหว่างรันไทม์ของเอเจนต์ในหน้าต่างด้านซ้าย

8ed90debb774207f.png

มาลองใช้การดำเนินการกัน แชทกับตัวแทนโดยใช้ข้อความแจ้งต่อไปนี้

  • " โปรดแสดงรายการ PRD ที่พร้อมใช้งานทั้งหมด "
  • " เขียนแผนการทดสอบสำหรับ PRD ของ Snaprecipe "

เมื่อใช้เครื่องมือบางอย่าง คุณจะตรวจสอบสิ่งที่เกิดขึ้นใน UI การพัฒนาได้

3469a0785b507be3.png

ดูว่าเอเจนต์ตอบกลับคุณอย่างไร และตรวจสอบเมื่อเราแจ้งให้ใช้ไฟล์ทดสอบ ซึ่งจะสร้างแผนการทดสอบในไฟล์ CSV เป็นอาร์ติแฟกต์

94b73fb4d447ad62.png

ตอนนี้คุณสามารถตรวจสอบเนื้อหาของ CSV ได้โดยการนำเข้าไปยัง Google ชีต เป็นต้น

89e3fc18d6e92d82.png

ยินดีด้วย ตอนนี้คุณมีเอเจนต์เครื่องมือวางแผนการทดสอบ QA ที่ใช้งานได้ซึ่งทำงานในเครื่องแล้ว ตอนนี้มาดูวิธีทำให้ใช้งานได้ใน Cloud Run เพื่อให้ผู้อื่นใช้ได้ด้วยกัน

5. การทำให้ใช้งานได้กับ Cloud Run

แน่นอนว่าตอนนี้เราต้องการเข้าถึงแอปที่ยอดเยี่ยมนี้จากทุกที่ หากต้องการทำเช่นนั้น เราสามารถแพ็กเกจแอปพลิเคชันนี้และทำให้ใช้งานได้กับ Cloud Run ในการสาธิตนี้ เราจะแสดงบริการนี้เป็นบริการสาธารณะที่ผู้อื่นเข้าถึงได้ อย่างไรก็ตาม โปรดทราบว่านี่ไม่ใช่แนวทางปฏิบัติแนะนำ

819179179d974c07.jpeg

ในไดเรกทอรีการทำงานปัจจุบัน เรามีไฟล์ทั้งหมดที่จำเป็นต่อการทำให้แอปพลิเคชันใช้งานได้ใน Cloud Run อยู่แล้ว ซึ่งได้แก่ ไดเรกทอรีของเอเจนต์, Dockerfile และ server.py (สคริปต์บริการหลัก) มาทำให้ใช้งานได้กันเลย ไปที่เทอร์มินัล Cloud Shell และตรวจสอบว่าได้กำหนดค่าโปรเจ็กต์ปัจจุบันเป็นโปรเจ็กต์ที่ใช้งานอยู่แล้ว หากยังไม่ได้กำหนดค่า ให้ใช้คำสั่ง gcloud configure เพื่อตั้งค่ารหัสโปรเจ็กต์

gcloud config set project [PROJECT_ID]

จากนั้นเรียกใช้คำสั่งต่อไปนี้เพื่อนำไปใช้งานใน Cloud Run

gcloud run deploy qa-test-planner-agent \
                  --source . \
                  --port 8080 \
                  --project {YOUR_PROJECT_ID} \
                  --allow-unauthenticated \
                  --region us-central1 \
                  --update-env-vars GOOGLE_GENAI_USE_VERTEXAI=1 \
                  --update-env-vars GOOGLE_CLOUD_PROJECT={YOUR_PROJECT_ID} \
                  --update-env-vars GOOGLE_CLOUD_LOCATION=global \
                  --update-env-vars CONFLUENCE_URL={YOUR_CONFLUENCE_URL} \
                  --update-env-vars CONFLUENCE_USERNAME={YOUR_CONFLUENCE_USERNAME} \
                  --update-env-vars CONFLUENCE_TOKEN={YOUR_CONFLUENCE_TOKEN} \
                  --update-env-vars CONFLUENCE_PRD_SPACE_ID={YOUR_PRD_SPACE_ID} \
                  --memory 1G

หากระบบแจ้งให้คุณรับทราบการสร้างรีจิสทรีของ Artifact สำหรับที่เก็บ Docker เพียงตอบว่า Y โปรดทราบว่าเราอนุญาตการเข้าถึงที่ไม่ได้ตรวจสอบสิทธิ์ที่นี่เนื่องจากนี่เป็นแอปพลิเคชันสาธิต เราขอแนะนำให้ใช้การตรวจสอบสิทธิ์ที่เหมาะสมสำหรับแอปพลิเคชันระดับองค์กรและการใช้งานจริง

เมื่อการติดตั้งใช้งานเสร็จสมบูรณ์แล้ว คุณจะได้รับลิงก์ที่คล้ายกับลิงก์ด้านล่าง

https://qa-test-planner-agent-*******.us-central1.run.app

เมื่อเข้าถึง URL คุณจะเข้าสู่ UI ของนักพัฒนาเว็บซึ่งคล้ายกับตอนที่คุณลองใช้ในเครื่อง คุณสามารถใช้แอปพลิเคชันจากหน้าต่างไม่ระบุตัวตนหรืออุปกรณ์เคลื่อนที่ได้เลย โดยควรจะพร้อมใช้งานแล้ว

ตอนนี้ลองใช้พรอมต์ต่างๆ เหล่านี้อีกครั้งตามลำดับ แล้วดูว่าจะเกิดอะไรขึ้น

  • " คุณค้นหา PRD ที่เกี่ยวข้องกับเครื่องมือประมาณการสินเชื่อได้ไหม "
  • "ช่วยแสดงความคิดเห็นเกี่ยวกับสิ่งที่เราควรปรับปรุงในเรื่องนี้"
  • " เขียนแผนการทดสอบสำหรับฟีเจอร์นี้"

นอกจากนี้ เนื่องจากเราเรียกใช้เอเจนต์เป็นแอป FastAPI เราจึงตรวจสอบเส้นทาง API ทั้งหมดในเส้นทาง /docs ได้ด้วย เช่น หากเข้าถึง URL ดังนี้ https://qa-test-planner-agent-*******.us-central1.run.app/docs คุณจะเห็นหน้าเอกสารประกอบ Swagger ดังที่แสดงด้านล่าง

c6f613b7bdc91ef3.png

คำอธิบายโค้ด

ตอนนี้มาดูไฟล์ที่เราต้องใช้ในการติดตั้งใช้งานกัน โดยเริ่มจาก server.py

# server.py

import os

from fastapi import FastAPI
from google.adk.cli.fast_api import get_fast_api_app

AGENT_DIR = os.path.dirname(os.path.abspath(__file__))

app_args = {"agents_dir": AGENT_DIR, "web": True}

app: FastAPI = get_fast_api_app(**app_args)

app.title = "qa-test-planner-agent"
app.description = "API for interacting with the Agent qa-test-planner-agent"


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8080)

เราสามารถแปลงเอเจนต์เป็นแอป FastAPI ได้อย่างง่ายดายโดยใช้ฟังก์ชัน get_fast_api_app ในฟังก์ชันนี้ เราสามารถตั้งค่าฟังก์ชันต่างๆ ได้ เช่น การกำหนดค่าบริการเซสชัน บริการอาร์ติแฟกต์ หรือแม้แต่การติดตามข้อมูลไปยังระบบคลาวด์

นอกจากนี้ คุณยังตั้งค่าวงจรแอปพลิเคชันได้ที่นี่ด้วยหากต้องการ หลังจากนั้นเราจะใช้ Uvicorn เพื่อเรียกใช้แอปพลิเคชัน Fast API ได้

หลังจากนั้น Dockerfile จะแสดงขั้นตอนที่จำเป็นในการเรียกใช้แอปพลิเคชัน

# Dockerfile

FROM python:3.12-slim

RUN pip install --no-cache-dir uv==0.7.13

WORKDIR /app

COPY . .

RUN uv sync --frozen

EXPOSE 8080

CMD ["uv", "run", "uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8080"]

6. ความท้าทาย

ตอนนี้ถึงเวลาที่คุณจะเฉิดฉายและขัดเกลาทักษะการสำรวจแล้ว คุณช่วยสร้างเครื่องมือเพื่อให้ระบบเขียนความคิดเห็นในการตรวจสอบ PRD ลงในไฟล์ด้วยได้ไหม

7. ล้างข้อมูล

โปรดทำตามขั้นตอนต่อไปนี้เพื่อเลี่ยงไม่ให้เกิดการเรียกเก็บเงินกับบัญชี Google Cloud สำหรับทรัพยากรที่ใช้ในโค้ดแล็บนี้

  1. ในคอนโซล Google Cloud ให้ไปที่หน้าจัดการทรัพยากร
  2. ในรายการโปรเจ็กต์ ให้เลือกโปรเจ็กต์ที่ต้องการลบ แล้วคลิกลบ
  3. ในกล่องโต้ตอบ ให้พิมพ์รหัสโปรเจ็กต์ แล้วคลิกปิดเพื่อลบโปรเจ็กต์
  4. หรือจะไปที่ Cloud Run ในคอนโซล เลือกบริการที่คุณเพิ่งทําให้ใช้งานได้ แล้วลบก็ได้