2. Google ADK, Custom Agent - 보안 게이트키퍼 시스템

사용자가 입력한 암호를 검증한 뒤, 인증에 성공했을 때만 기밀 정보를 처리하는 AI 비서 에이전트로 제어권(Handoff/Transfer)을 넘겨주는 보안 게이트키퍼 시스템입니다.

코드의 구성을 1) 모듈 임포트, 2) 작업 에이전트 정의, 3) 커스텀 게이트키퍼 클래스, 4) 인스턴스 생성 및 Root Agent 설정 영역으로 나누어 한 줄 한 줄 상세히 설명해 드리겠습니다.

1. 모듈 임포트 (Import) 영역

import os
from typing import AsyncGenerator

from google.adk import Agent
from google.adk.agents import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event, EventActions
from google.genai.types import Content, Part

• import os: 시스템 환경변수에 접근하기 위한 Python 표준 라이브러리입니다. GOOGLE_MODEL 같은 키값을 불러올 때 사용합니다.
• from typing import AsyncGenerator: 비동기 제너레이터의 반환 타입을 지정하기 위한 타입 힌팅 도구입니다. 이벤트를 지속적으로 생성(yield)하여 내보내는 비동기 함수에 사용됩니다.
• from google.adk import Agent: Google ADK의 기본 LLM 에이전트 클래스입니다. 프롬프트 지시사항(Instruction)을 받아 LLM 기반으로 대화하는 에이전트를 만들 때 사용합니다.
• from google.adk.agents import BaseAgent: ADK의 모든 에이전트가 상속받는 최상위 추상 클래스입니다. LLM에 의존하지 않고 Python 코드로 직접 로직(조건문, 비밀번호 검증 등)을 제어하는 커스텀 에이전트를 만들 때 상속받습니다.
• from google.adk.agents.invocation_context import InvocationContext: 에이전트가 실행되는 시점의 대화 맥락, 사용자 입력 메시지, 이전 세션 정보 등을 담고 있는 실행 컨텍스트 객체입니다.
• from google.adk.events import Event, EventActions: ADK 엔진과 에이전트 간에 주고받는 이벤트(메시지 출력, 에이전트 전환 요청 등)를 정의하는 객체입니다.
• from google.genai.types import Content, Part: Google GenAI 데이터 규격에 맞게 텍스트 콘텐츠 메시지를 구성하기 위한 클래스입니다.

 

2. 기밀 정보 담당 비서 에이전트 (worker_agent)

worker_agent = Agent(
    name="secret_assistant",
    model=os.getenv("GOOGLE MODEL", "gemini-3.5-flash"),
    description="최고 보안 등급 기밀 정보 처리 비서",
    instruction="당신은 기밀 정보를 다루는 최고 보안 등급의 AI 비서입니다. 사용자 질문에 친절하고 정밀하게 답하세요."
)

• name="secret_assistant": 에이전트의 식별자 이름입니다. 추후 게이트키퍼가 이 이름을 통해 제어권을 넘겨주게(Transfer) 됩니다.
• model=os.getenv(...): 환경변수에서 모델명을 가져오며, 없을 경우 기본값으로 지정된 Gemini 모델을 사용합니다.
• description / instruction: 이 에이전트의 역할 설명 및 페르소나/행동 지침 프롬프트입니다.

 

3. 커스텀 보안 게이트키퍼 클래스 (GateKeeperAgent)

class GateKeeperAgent(BaseAgent):
    secret_password: str = "열려라 참깨"
    target_agent_name: str = "secret_assistant" 

• class GateKeeperAgent(BaseAgent):: BaseAgent를 상속받아 직접 비동기 로직(_run_async_impl)을 구현할 수 있는 커스텀 에이전트를 정의합니다.
• secret_password / target_agent_name: 클래스의 기본 속성으로, 성공시 비교할 비밀번호 문자열과 통과 후 전환할 대상 에이전트의 이름을 지정합니다.

비동기 실행 로직 함수 (_run_async_impl)

    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:

• _run_async_impl: ADK 에이전트가 호출될 때 실행되는 메인 엔진 메서드입니다. InvocationContext를 통해 들어온 메시지를 처리하고 Event를 생성하여 반환합니다.

        user_input = ""        
        if ctx.user_content and ctx.user_content.parts:
            first_part = ctx.user_content.parts[0]
            if hasattr(first_part, "text") and first_part.text:
                user_input = first_part.text.strip()

• 사용자 입력 추출: ctx.user_content에서 사용자가 방금 입력한 텍스트 파트(parts[0])를 안전하게 추출하고, 양끝 공백을 제거(.strip())하여 user_input 변수에 저장합니다.

비밀번호 검증 및 이벤트 발송 (성공 시)

        if user_input == self.secret_password:
            # 1) 성공 메시지 전송
            yield Event(
                content=Content(parts=[Part.from_text(text="보안인증 성공! 비서에게 연결합니다. 질문해 주세요.")]),
                author=self.name,
            )

• 비밀번호 일치 확인: 사용자의 입력이 "열려라 참깨"와 일치하는지 확인합니다.
• 안내 메시지 발송 (yield Event): 인증 성공 메시지를 사용자 화면에 출력합니다. author=self.name은 이 메시지의 작성자가 게이트키퍼임을 밝힙니다.

            # 2) target_agent_name으로 제어권 핸드오프 (Transfer)
            yield Event(
                actions=EventActions(transfer_to_agent=self.target_agent_name),
                author=self.name
            )

• 에이전트 전환 요청: EventActions(transfer_to_agent="secret_assistant")를 포함한 이벤트를 내보냅니다. 이를 받은 ADK 프레임워크는 대화 세션의 주도권을 즉시 worker_agent(secret_assistant)에게 넘깁니다.

비밀번호 검증 및 이벤트 발송 (실패 시)

        else:
            # 실패 메시지 전송
            yield Event(
                content=Content(parts=[Part.from_text(text="접근 거부: 비밀번호가 틀렸습니다. 올바른 암호를 입력하세요.")]),
                author=self.name,
            )

• 인증 실패 처리: 암호가 틀렸을 경우 거부 메시지를 출력하며, transfer_to_agent를 호출하지 않아 제어권을 넘기지 않고 계속 게이트키퍼에 머물게 합니다.

4. 인스턴스 생성 및 Root Agent 래핑

gatekeeper = GateKeeperAgent(
    name="security_gate",
    secret_password="열려라 참깨",
    sub_agents=[worker_agent]
)

• gatekeeper 객체 생성: 보안 관문 역할을 할 GateKeeperAgent 인스턴스를 생성합니다. sub_agents=[worker_agent]를 통해 자신이 제어권을 전환할 수 있는 하위 에이전트로 worker_agent를 등록해 둡니다.

root_agent = Agent(
    name="security_system",
    model=os.getenv("GOOGLE MODEL", "gemini-3.5-flash"),
    description="보안 관문 제어 시스템",
    instruction="사용자 입력을 게이트키퍼로 전달하여 인증을 수행하도록 하세요.",
    sub_agents=[gatekeeper]
)

• root_agent (최상위 에이전트): ADK Dev UI 호환성 및 라우트 생성을 위해 최상위를 일반 Agent로 두고 sub_agents=[gatekeeper]로 래핑했습니다.
• 전체 흐름은 root_agent ->  gatekeeper (비밀번호 검증) -> (성공 시) worker_agent로 완벽히 제어됩니다.

 

    import os

    from typing import AsyncGenerator

    from google.adk import Agent

    from google.adk.agents import BaseAgent

    from google.adk.agents.invocation_context import InvocationContext

    from google.adk.events import Event, EventActions

    from google.genai.types import Content, Part

    # 1. 기밀 정보를 담당할 대상 에이전트 (Worker Agent)

    worker_agent = Agent(

        name="secret_assistant",

        model=os.getenv("GOOGLE MODEL", "gemini-3.5-flash"),

        description="최고 보안 등급 기밀 정보 처리 비서",

        instruction="당신은 기밀 정보를 다루는 최고 보안 등급의 AI 비서입니다. 사용자 질문에 친절하고 정밀하게 답하세요."

    )

    # 2. 커스텀 보안 게이트키퍼 에이전트 (GateKeeper Agent)

    class GateKeeperAgent(BaseAgent):

        secret_password:str="이게 비번"

        target_agent_name:str = "secret_assistant"  

       

        async def _run_async_impl(self, ctx:InvocationContext)->AsyncGenerator[Event, None]:

            user_input=""        

           

            # 최신 InvocationContext 내 user_content 안전 추출    

            if ctx.user_content and ctx.user_content.parts:

                first_part = ctx.user_content.parts[0]

                if hasattr(first_part, "text") and first_part.text:

                    user_input = first_part.text.strip()

               

            # 비밀번호 검증 조건문            

            if user_input == self.secret_password:

                # 1) 성공 메시지 전송

                yield Event(

                    content=Content(parts=[Part.from_text(text="보안인증 성공! 비서에게 연결합니다. 질문해 주세요.")]),

                    author=self.name,

                )

               

                # 2) target_agent_name으로 제어권 핸드오프 (Transfer)

                yield Event(

                    actions=EventActions(transfer_to_agent=self.target_agent_name),

                    author=self.name

                )

            else:

                # 실패 메시지 전송

                yield Event(

                    content=Content(parts=[Part.from_text(text="접근 거부: 비밀번호가 틀렸습니다. 올바른 암호를 입력하세요.")]),

                    author=self.name,

                )

    #3. 게이트키퍼 에이전트 인스턴스 생성

    gatekeeper = GateKeeperAgent(

        name="security_gate",

        secret_password="이게 비번",

        sub_agents=[worker_agent]

    )

    # root_agent = gatekeeper

    root_agent = Agent(

        name="security_system",

        model=os.getenv("GOOGLE MODEL", "gemini-3.5-flash"),

        description="보안 관문 제어 시스템",

        instruction="사용자 입력을 게이트기퍼로 전달하여 인증을 수행하도록 하세요.",

        sub_agents=[gatekeeper]

    )