AI Agent 핵심 프로토콜 완전 정복: MCP부터 AG-UI까지

from google.adk.agents import Agent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from google.adk.tools.toolbox_toolset import ToolboxToolset
from mcp import StdioServerParameters

# [Database MCP] 창고 재고(소고기, 연어 등)를 조회하는 도구
inventory_tools = ToolboxToolset(server_url=TOOLBOX_URL)

# [Notion MCP] 레시피 및 거래처 연락처를 조회하는 도구
notion_tools = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@notionhq/notion-mcp-server"],
            env={"NOTION_TOKEN": NOTION_TOKEN}
        ),
        timeout=30
    )
)

# [Mailgun MCP] 부족한 재고를 채우기 위해 공급업체에 이메일을 보내는 도구
mailgun_tools = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@mailgun/mcp-server"],
            env={"MAILGUN_API_KEY": MAILGUN_API_KEY}
        ),
        timeout=30
    )
)

# [에이전트 생성] 이름, 모델, 지시 사항, 사용할 도구를 조합해 에이전트를 만든다
kitchen_agent = Agent(
    model="gemini-3-flash-preview",
    name="kitchen_manager",
    instruction="당신은 식당 주방을 관리합니다. 재고를 확인하고, 레시피를 찾아보며, 공급업체에 이메일을 보내세요.",
    tools=[inventory_tools, notion_tools, mailgun_tools],
)

import asyncio
from google.adk.agents import Agent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from google.adk.tools.toolbox_toolset import ToolboxToolset
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from a2a.client.client_factory import ClientFactory
from a2a.client.helpers import create_text_message_object
from mcp import StdioServerParameters

# ── 1단계: 에이전트가 사용할 MCP 도구 설정 ────────────────────────────────

inventory_tools = ToolboxToolset(server_url=TOOLBOX_URL)

notion_tools = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@notionhq/notion-mcp-server"],
            env={"NOTION_TOKEN": NOTION_TOKEN}
        ),
        timeout=30
    )
)

mailgun_tools = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@mailgun/mcp-server"],
            env={"MAILGUN_API_KEY": MAILGUN_API_KEY}
        ),
        timeout=30
    )
)

# ── 2단계: 에이전트 생성 및 A2A 서비스로 노출 ─────────────────────────────

kitchen_agent = Agent(
    model="gemini-3-flash-preview",
    name="kitchen_manager",
    instruction="당신은 식당 주방을 관리합니다. 재고를 확인하고, 레시피를 찾아보며, 공급업체에 이메일을 보내세요.",
    tools=[inventory_tools, notion_tools, mailgun_tools]
)

# 8080 포트로 에이전트를 A2A 서비스로 노출한다.
# 외부 에이전트는 http://localhost:8080/.well-known/agent-card.json 에서 내 정보를 확인한다.
app = to_a2a(kitchen_agent, port=8080)

# ── 3단계: 원격 에이전트(A2A)와 통신 ─────────────────────────────────────

async def check_wholesale_prices():
    # 가격 조회 전문 에이전트에 연결
    client = await ClientFactory.connect("http://pricing-agent:8001")

    # 연결된 에이전트의 Agent Card 정보를 확인
    card = await client.get_card()
    print(f"연결 성공: {card.name}")

    # 도매가 정보를 요청하고 응답을 스트리밍으로 수신
    message = create_text_message_object(content="오늘 연어 도매가 알려줘")
    async for response in client.send_message(message):
        pass

asyncio.run(check_wholesale_prices())

import asyncio
import httpx
import uuid
from ucp_sdk.models.discovery.profile_schema import UcpDiscoveryProfile
from ucp_sdk.models.schemas.shopping.checkout_create_req import CheckoutCreateRequest
from ucp_sdk.models.schemas.shopping.types.line_item_create_req import LineItemCreateRequest
from ucp_sdk.models.schemas.shopping.types.item_create_req import ItemCreateRequest
from ucp_sdk.models.schemas.shopping.payment_create_req import PaymentCreateRequest

# 1. 공급업체의 UCP 상거래 규격 조회
async def get_ucp_profile(base_url: str) -> UcpDiscoveryProfile:
    async with httpx.AsyncClient() as c:
        res = await c.get(f"{base_url}/.well-known/ucp")
        return UcpDiscoveryProfile.model_validate(res.json())

# 2. 아이템 딕셔너리를 UCP 표준 장바구니 객체로 변환
def create_ucp_checkout_request(items: dict, currency: str = "USD") -> CheckoutCreateRequest:
    line_items = [
        LineItemCreateRequest(quantity=qty, item=ItemCreateRequest(id=item_id))
        for item_id, qty in items.items()
    ]
    return CheckoutCreateRequest(
        currency=currency,
        line_items=line_items,
        payment=PaymentCreateRequest()
    )

# 3. 에이전트 신원 및 중복 결제 방지용 헤더 생성
def create_ucp_headers() -> dict:
    return {
        "UCP-Agent": 'profile="https://kitchen.example/agent"',
        "Idempotency-Key": str(uuid.uuid4()),
        "Request-Id": str(uuid.uuid4())
    }

# 4. 체크아웃 세션 생성
async def create_checkout_session(base_url: str, checkout_req: CheckoutCreateRequest, headers: dict) -> dict:
    async with httpx.AsyncClient() as c:
        res = await c.post(
            f"{base_url}/checkout-sessions",
            json=checkout_req.model_dump(mode="json", by_alias=True, exclude_none=True),
            headers=headers
        )
        return res.json()

# 5. 주문 최종 완료
async def complete_ucp_order(base_url: str, checkout_id: str, headers: dict):
    headers["Idempotency-Key"] = str(uuid.uuid4())  # 완료 요청은 새 키로
    async with httpx.AsyncClient() as c:
        return await c.post(
            f"{base_url}/checkout-sessions/{checkout_id}/complete",
            headers=headers
        )

# ── 통합 실행 예시 ──────────────────────────────────────────────────────────

async def run_ucp_order():
    supplier_url = "http://example-wholesale:8182"
    order_items = {"salmon": 10, "olive_oil": 3}

    await get_ucp_profile(supplier_url)
    checkout_req = create_ucp_checkout_request(order_items)
    headers = create_ucp_headers()
    session = await create_checkout_session(supplier_url, checkout_req, headers)
    result = await complete_ucp_order(supplier_url, session["id"], headers)
    print(f"주문 완료: {result.status_code}")

asyncio.run(run_ucp_order())

from google.adk.agents import Agent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from google.adk.tools.toolbox_toolset import ToolboxToolset
from mcp import StdioServerParameters

INVENTORY_SERVER_URL = "TOOLBOX_URL"
NOTION_API_TOKEN = "YOUR_TOKEN"
MAILGUN_API_KEY = "YOUR_KEY"

# [Database MCP] 창고 재고를 조회하는 도구
inventory_tools = ToolboxToolset(server_url=INVENTORY_SERVER_URL)

# [Notion MCP] 레시피 및 거래처 연락처를 조회하는 도구
notion_tools = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@notionhq/notion-mcp-server"],
            env={"NOTION_TOKEN": NOTION_API_TOKEN}
        ),
        timeout=30
    )
)

# [Mailgun MCP] 공급업체에 이메일을 보내는 도구
mailgun_tools = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@mailgun/mcp-server"],
            env={"MAILGUN_API_KEY": MAILGUN_API_KEY}
        ),
        timeout=30
    )
)

# [에이전트 본체] 페르소나와 도구를 결합
kitchen_agent = Agent(
    model="gemini-3-flash-preview",
    name="kitchen_manager",
    instruction="당신은 식당 주방을 관리합니다. 재고를 확인하고 주문을 진행하세요.",
    tools=[inventory_tools, notion_tools, mailgun_tools],
)

import httpx
import uuid
from a2a.client.client_factory import ClientFactory
from ucp_sdk.models.discovery.profile_schema import UcpDiscoveryProfile
from ucp_sdk.models.schemas.shopping.checkout_create_req import CheckoutCreateRequest
from ucp_sdk.models.schemas.shopping.types.line_item_create_req import LineItemCreateRequest
from ucp_sdk.models.schemas.shopping.types.item_create_req import ItemCreateRequest
from ucp_sdk.models.schemas.shopping.payment_create_req import PaymentCreateRequest
from ap2.types.mandate import IntentMandate, PaymentMandate, PaymentMandateContents
from ap2.types.payment_request import PaymentCurrencyAmount, PaymentItem, PaymentResponse
from ap2.types.payment_receipt import PaymentReceipt, Success

# ── A2A ────────────────────────────────────────────────────────────────────

# 상대 에이전트의 주소로 통신 채널을 연결
async def discover_supplier_agent(url: str):
    return await ClientFactory.connect(url)

# ── UCP ────────────────────────────────────────────────────────────────────

# 공급업체의 상거래 규격 조회
async def get_ucp_profile(base_url: str) -> UcpDiscoveryProfile:
    async with httpx.AsyncClient() as c:
        res = await c.get(f"{base_url}/.well-known/ucp")
        return UcpDiscoveryProfile.model_validate(res.json())

# 아이템 딕셔너리를 UCP 표준 장바구니 객체로 변환
def create_ucp_checkout_request(items: dict, currency: str = "USD") -> CheckoutCreateRequest:
    line_items = [
        LineItemCreateRequest(quantity=qty, item=ItemCreateRequest(id=i))
        for i, qty in items.items()
    ]
    return CheckoutCreateRequest(
        currency=currency,
        line_items=line_items,
        payment=PaymentCreateRequest()
    )

# 에이전트 신원 및 중복 결제 방지용 헤더 생성
def create_ucp_headers() -> dict:
    return {
        "UCP-Agent": 'profile="https://kitchen.example/agent"',
        "Idempotency-Key": str(uuid.uuid4()),
        "Request-Id": str(uuid.uuid4())
    }

# ── AP2 ────────────────────────────────────────────────────────────────────

# 에이전트가 사용할 지출 한도와 허용 업체를 사전 정의
def setup_payment_guardrails() -> IntentMandate:
    return IntentMandate(
        natural_language_description="Order salmon & olive oil",
        merchants=["Example Wholesale"],
        intent_expiry="2026-03-21T20:00:00Z"
    )

# 특정 주문에 대해 사용자 권한을 위임한 결제 승인서 생성
def create_payment_mandate(order_id: str, total_value: float, merchant_name: str) -> PaymentMandate:
    contents = PaymentMandateContents(
        payment_mandate_id=str(uuid.uuid4())[:8],
        payment_details_id=order_id,
        payment_details_total=PaymentItem(
            label="Kitchen Order",
            amount=PaymentCurrencyAmount(currency="USD", value=total_value)
        ),
        payment_response=PaymentResponse(request_id=order_id, method_name="CARD"),
        merchant_agent=merchant_name
    )
    mandate = PaymentMandate(payment_mandate_contents=contents)
    mandate.user_authorization = f"signed_{contents.payment_mandate_id}"
    return mandate

# 결제 완료 후 감사용 영수증 발행
def create_payment_receipt(mandate_id: str, total_value: float, confirmation_id: str) -> PaymentReceipt:
    return PaymentReceipt(
        payment_mandate_id=mandate_id,
        payment_id=f"PAY-{str(uuid.uuid4())[:8]}",
        amount=PaymentCurrencyAmount(currency="USD", value=total_value),
        payment_status=Success(merchant_confirmation_id=confirmation_id),
    )

import asyncio
import httpx
import uuid
from config import kitchen_agent
from protocols import (
    discover_supplier_agent,
    get_ucp_profile,
    create_ucp_checkout_request,
    create_ucp_headers,
    setup_payment_guardrails,
    create_payment_mandate,
    create_payment_receipt,
)

async def run_kitchen_workflow():
    order_items = {"salmon": 10, "olive_oil": 3}
    supplier_url = "http://example-wholesale:8182"
    total_price = 294.00

    # 1. [AP2] 지출 한도와 허용 업체를 먼저 설정
    intent = setup_payment_guardrails()

    # 2. [A2A] 가격 조회 전문 에이전트에 연결
    await discover_supplier_agent("http://pricing-agent:8001")

    # 3. [UCP] 공급업체의 상거래 규격 확인
    await get_ucp_profile(supplier_url)

    # 4. [UCP] 표준 장바구니 객체 및 헤더 생성
    checkout_req = create_ucp_checkout_request(order_items)
    headers = create_ucp_headers()

    # 5. [UCP] 체크아웃 세션 생성
    async with httpx.AsyncClient() as c:
        res = await c.post(
            f"{supplier_url}/checkout-sessions",
            json=checkout_req.model_dump(mode="json", by_alias=True, exclude_none=True),
            headers=headers
        )
        session = res.json()

    # 6. [AP2] 결제 승인서 생성 및 헤더에 첨부
    mandate = create_payment_mandate(session["id"], total_price, intent.merchants[0])
    headers["Idempotency-Key"] = str(uuid.uuid4())
    headers["Authorization-Mandate"] = mandate.user_authorization

    # 7. [UCP + AP2] 승인서가 포함된 최종 주문 완료
    async with httpx.AsyncClient() as c:
        final_res = await c.post(
            f"{supplier_url}/checkout-sessions/{session['id']}/complete",
            headers=headers
        )

        if final_res.status_code == 200:
            # 8. [AP2] 감사용 영수증 발행
            receipt = create_payment_receipt(
                mandate.payment_mandate_contents.payment_mandate_id,
                total_price,
                final_res.json().get("confirmation_id", "CONF-001")
            )
            print(f"결제 완료. 영수증 ID: {receipt.payment_id}")
        else:
            print(f"결제 실패: {final_res.status_code}")

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

# 에이전트가 전송하는 A2UI 데이터 구조
# 클라이언트 측 렌더러가 이를 받아 네이티브 UI로 변환한다

a2ui_messages = [
    # 1. 렌더링을 시작할 화면(Surface) 생성
    {"beginRendering": {"surfaceId": "default", "root": "card"}},

    # 2. 컴포넌트 트리 전송: 중첩 구조가 아닌 ID 참조 방식의 평면 리스트
    {"surfaceUpdate": {"surfaceId": "default", "components": [
        # 'card'는 최상위 카드 컨테이너, 자식으로 'col'을 가짐
        {"id": "card",      "component": {"Card": {"child": "col"}}},

        # 'col'은 세로 정렬 컨테이너, 제목·가격·버튼을 순서대로 배치
        {"id": "col",       "component": {"Column": {"children": {"explicitList": ["title", "price", "buy"]}}}},

        # 'title'은 H3 스타일 텍스트, 내용은 데이터 경로 "name"에서 바인딩
        {"id": "title",     "component": {"Text": {"usageHint": "h3", "text": {"path": "name"}}}},

        # 'price'는 일반 텍스트, 내용은 데이터 경로 "price"에서 바인딩
        {"id": "price",     "component": {"Text": {"text": {"path": "price"}}}},

        # 'buy'는 클릭 시 "purchase" 액션을 실행하는 버튼
        {"id": "buy",       "component": {"Button": {"child": "btn-label",
            "action": {"name": "purchase", "context": [{"key": "item", "value": {"path": "name"}}]}}}},

        # 'btn-label'은 버튼 내부의 고정 텍스트
        {"id": "btn-label", "component": {"Text": {"text": {"literalString": "Buy Now"}}}},
    ]}},

    # 3. 실제 데이터 전송: 구조 재전송 없이 데이터만 업데이트 가능
    {"dataModelUpdate": {"surfaceId": "default", "contents": [
        {"key": "name",  "valueString": "Fresh Atlantic Salmon"},
        {"key": "price", "valueString": "$24.00/lb"},
    ]}},
]

from ag_ui_adk import ADKAgent, add_adk_fastapi_endpoint
from fastapi import FastAPI

# 에이전트를 AG-UI 규격으로 감싸고, FastAPI 앱에 스트리밍 엔드포인트를 추가
ag_ui_agent = ADKAgent(adk_agent=kitchen_agent, app_name="kitchen", user_id="chef")
app = FastAPI()
add_adk_fastapi_endpoint(app, ag_ui_agent, path="/")

# 실행: uvicorn main:app

# ── 클라이언트가 수신하는 표준 SSE 이벤트 스트림 ──────────────────────────
#
# RUN_STARTED           에이전트가 요청을 받아 추론을 시작
# TOOL_CALL_START       도구 호출 시작  (예: toolCallName="check_inventory")
# TOOL_CALL_RESULT      도구 실행 결과  (예: content="재고 3파운드, 재주문 필요")
# TOOL_CALL_END         도구 호출 프로세스 종료
# TEXT_MESSAGE_CONTENT  답변 생성 중    (델타: "현재 재고 상황을 바탕으로...")
# TEXT_MESSAGE_CONTENT  답변 생성 중    (델타: "연어 10파운드 주문을 권장합니다.")
# RUN_FINISHED          모든 응답 프로세스 완료