확정적 MCP 호출 (deterministic_tool)

개요

@deterministic_tool은 ReAct 루프가 시작되기 전에 미리 실행되는 분기 규칙입니다. 함수가 dict를 반환하면 지정한 MCP tool을 즉시 호출하고, None을 반환하면 건너뛰어 ReAct가 평소처럼 판단합니다.

사용자 요청 → input guardrail
           ↓
    [ Deterministic 단계 ]  ← 여기서 확정 호출
           ↓
      ReAct think/act loop
           ↓
       output guardrail → 응답

언제 쓰는가

시나리오	왜 필요한가
명확한 의도 라우팅	”주문 조회” 질의에 항상 `get_order` 호출 — LLM이 비슷한 tool을 고를 위험 제거
latency/cost 최적화	LLM round-trip 한 번 제거 → 수백 ms·토큰 절감
규제/감사 요건	금융·의료 도메인에서 “어떤 입력에 어떤 tool을 썼는지” 감사 로그 확정
구조화 입력 분기	A2A `DataPart`의 `chatbotId` 등 메타데이터로 분기

반대로 LLM의 자율 판단이 가치 있는 일반 대화형 에이전트라면 @tool 또는 MCP만으로 충분합니다. deterministic은 “예측 가능성이 품질”인 경우에 쓰세요.

동작 방식

사용자 요청 수신 → 입력 guardrail 통과
DETERMINISTIC_TOOLS 리스트를 등록 순서대로 순회
- 함수가 dict 반환 → 해당 MCP tool 즉시 호출, 결과 누적
- None 반환 → skip, 다음 함수 평가
실행된 모든 결과가 state["metadata"]["deterministic_results"]에 첨부
LLM system message에 결과가 자동 주입
ReAct 루프 시작 — LLM이 결과를 참고해 추가 tool 호출 여부를 판단

`DeterministicContext` API

라우팅 함수가 받는 유일한 인자입니다. dataclass이므로 테스트에서 직접 인스턴스화할 수 있습니다.

속성	타입	설명
`query`	`str`	사용자 입력 텍스트
`context`	`dict`	A2A context (`userId`, `sessionId`, `chatbotId` 등)
`a2a_data`	`list[dict] \| None`	A2A `DataPart` 전체
`a2a_files`	`list[dict] \| None`	A2A `FilePart` 메타
`mcp_tools`	`list[BaseTool]`	바인딩된 `mcp_id`에 속한 tool 목록 (자동 필터링됨)

규칙 작성 예시

예시 1: 키워드 기반 분기

from llamon_agent.tools import deterministic_tool, DeterministicContext


@deterministic_tool(mcp_id="order-service")
def route_order_lookup(ctx: DeterministicContext) -> dict | None:
    """'주문 조회' 키워드 감지 시 get_order tool을 확정 호출."""
    if "주문" in ctx.query and "조회" in ctx.query:
        return {
            "tool_name": "get_order",
            "params": {"query": ctx.query},
        }
    return None  # 매치 안 됨 → ReAct로 위임


DETERMINISTIC_TOOLS = [
    route_order_lookup,
]

예시 2: A2A `DataPart`로 분기

@deterministic_tool(mcp_id="customer-crm")
def route_by_chatbot_id(ctx: DeterministicContext) -> dict | None:
    """A2A context의 chatbotId에 따라 CRM lookup을 확정 호출."""
    chatbot_id = ctx.context.get("chatbotId")
    if chatbot_id == "premium-support":
        return {
            "tool_name": "get_customer_profile",
            "params": {"customer_id": ctx.context.get("userId")},
        }
    return None

예시 3: 사용 가능한 tool 동적 선택

@deterministic_tool(mcp_id="knowledge-base")
def route_by_available_tool(ctx: DeterministicContext) -> dict | None:
    """mcp_tools에서 이름으로 동적 선택 후 호출."""
    tool_names = {t.name for t in ctx.mcp_tools}
    if "search_docs" in tool_names and "검색" in ctx.query:
        return {"tool_name": "search_docs", "params": {"q": ctx.query}}
    return None

에러 처리

상황	동작
라우팅 함수 내 예외 발생	`"확정 호출된 MCP 단계에서 오류가..."` 응답 반환, LLM 호출 차단
반환 dict의 `tool_name`이 MCP에 없음	동일하게 실패 응답
`mcp_id`가 실제 바인딩에 없음	`"MCP '...'의 tool이 로드되지 않음"` 에러
`None` 반환	정상 skip (에러 아님)

ReAct 혼합 패턴

deterministic 결과는 LLM의 system message에 참고 자료로 주입되지만, LLM이 추가 tool을 호출할지는 자율적으로 판단합니다. 한 번의 deterministic 호출로 답변을 끝내려면 system prompt에 명시적 지시를 추가하세요:

제공된 "확정 호출 결과"가 있으면 그 내용을 우선 사용하여 답변하고,
추가 도구 호출은 필요한 경우에만 하세요.

반대로 deterministic을 **“선행 조회”**로만 쓰고 이후 ReAct가 후속 처리하도록 설계할 수도 있습니다 — 예시 2의 CRM lookup → LLM이 받은 고객 정보로 맞춤 답변 생성.

여러 규칙 순서

DETERMINISTIC_TOOLS는 등록 순서대로 평가됩니다
여러 규칙이 동시에 match → 모두 실행 후 결과가 누적 (각각 mcp_id가 달라도 OK)
같은 mcp_id에서 여러 tool 호출도 가능
서로 다른 mcp_id 간에는 병렬이 아닌 순차 실행 (SDK 내부 결정)

테스트 전략

DeterministicContext는 dataclass라 pytest에서 직접 생성해 unit 테스트 가능:

from llamon_agent.tools import DeterministicContext
from app.tools import route_order_lookup


def test_route_matches_on_keyword():
    ctx = DeterministicContext(
        query="주문 조회 부탁해",
        context={},
        a2a_data=None,
        a2a_files=None,
        mcp_tools=[],
    )
    result = route_order_lookup(ctx)
    assert result == {"tool_name": "get_order", "params": {"query": "주문 조회 부탁해"}}


def test_route_skips_when_no_match():
    ctx = DeterministicContext(
        query="날씨 어때?",
        context={}, a2a_data=None, a2a_files=None, mcp_tools=[],
    )
    assert route_order_lookup(ctx) is None

실제 MCP 통합 테스트는 llamon run . 후 서버 로그에서 "Deterministic pre-step 완료: N건" 로그로 확인합니다.

Flow 모드에서는?

Flow 템플릿(graph-sequential, graph-parallel, graph-conditional, graph-http-pipeline)에서는 @deterministic_tool을 사용하지 않습니다. 대신 노드 내부에서 MCPHandle을 사용합니다.

왜 Flow는 다른 메커니즘을 쓰는가

@deterministic_tool의 존재 이유는 **“ReAct라는 자율 루프에서 예측 가능성을 확보”**하기 위함입니다. Flow는 개발자가 그래프 노드를 명시적으로 짜므로 이 문제가 구조적으로 없습니다. 노드 자체가 이미 확정적인 분기입니다.

축	ReAct agent	Flow graph
실행 순서 결정	LLM이 동적 판단	개발자가 그래프로 명시
”확정 호출”이 필요한 이유	LLM이 잘못된 tool을 고를 수 있음	노드 자체가 확정 실행
기능명	`@deterministic_tool`	`MCPHandle.call()`
위치	`app/tools.py`	`app/nodes.py`
등록	`DETERMINISTIC_TOOLS` 리스트	모듈 레벨 `mcp = MCPHandle()`

`MCPHandle`로 노드에서 확정 호출

Flow 템플릿의 app/nodes.py는 이미 아래 패턴으로 생성됩니다:

# app/nodes.py (Flow scaffold 기본)
from llamon_agent import MCPHandle

mcp = MCPHandle()  # 모듈 레벨 1회 선언

from app.nodes import mcp

async def build_graph(settings):
    # 1회만 바인딩 — 이후 노드에서 mcp.call() 사용 가능
    mcp_id = resolve_env_override("MCP_ID", "<FALLBACK>", source_file=__file__)
    await mcp.bind_registry(settings, mcp_ids=[mcp_id], mcp_id=mcp_id)
    ...

# app/nodes.py — 노드에서 확정 호출
from app.nodes import mcp

async def business_node(state) -> dict:
    # 노드 조건에 따라 원하는 tool을 직접 호출
    result = await mcp.call("get_order", params={"query": state["query"]})
    return {"output": result, "messages": [...]}

`MCPHandle` API

메서드	설명
`mcp = MCPHandle()`	모듈 레벨 선언 (싱글톤처럼 사용)
`await mcp.bind_registry(settings, mcp_ids=[...], mcp_id=...)`	그래프 빌드 시점 1회 바인딩 (Registry 모드)
`await mcp.call(tool_name, params={...})`	tool 실행. 결과는 dict/string 반환

여러 MCP를 사용하려면 handle을 분리하세요:

crm_mcp = MCPHandle()
web_mcp = MCPHandle()

# build_graph에서 각각 bind_registry
await crm_mcp.bind_registry(settings, mcp_ids=[crm_id], mcp_id=crm_id)
await web_mcp.bind_registry(settings, mcp_ids=[web_id], mcp_id=web_id)

# 노드에서 용도에 맞게 선택
async def lookup_node(state):
    customer = await crm_mcp.call("get_customer", params={"id": state["user_id"]})
    ...

대응 비교 — 언제 무엇을 쓰나

요구사항	ReAct 템플릿	Flow 템플릿
”조회” 키워드 → `get_order` 확정 호출	`@deterministic_tool`로 규칙 등록	노드의 `if "조회" in state["query"]` 조건문 + `mcp.call()`
순차 MCP 체인 (A → B → C)	deterministic 3개 등록 (혹은 ReAct에 맡김)	노드를 A/B/C로 나누고 `.edge()` 연결
사용자 입력에 따라 다른 tool	`DETERMINISTIC_TOOLS` 순회 + 첫 매치 사용	`edge_conditional()`로 분기

원칙: Flow에서는 “조건문”이 @deterministic_tool을 대체합니다. 그래프 구조가 이미 확정 흐름을 표현하기 때문입니다.

문제 해결

증상	원인/해결
함수가 아예 호출되지 않음	`app/tools.py`의 `DETERMINISTIC_TOOLS` 대상에 포함됐는지 확인
`"MCP 'xxx'의 tool이 로드되지 않음"`	Registry에 해당 MCP가 바인딩되었는지, `mcp_id` 오타 여부 확인
`"확정 호출된 MCP 단계에서 오류..."`	라우팅 함수 예외 또는 `tool_name` 오타. 서버 로그에서 trace 확인
LLM이 deterministic 결과를 무시하고 다시 같은 tool 호출	system prompt에 “확정 호출 결과를 우선 사용” 명시
Flow 노드에서 `deterministic_results` 읽는 코드 추천	Flow에서는 사용하지 마세요. `MCPHandle`을 쓰세요 (이 페이지 위 섹션 참고)