도구를 쥐여 주기: Tool과 MCP

 1# tools_basic.py - @tool 데코레이터 3단계 깊이로 써보기
 2from langchain_core.tools import tool
 3from pydantic import BaseModel, Field
 4
 5
 6# 단계 1: 가장 단순한 형태
 7# 이름은 함수 이름, 설명은 docstring, 스키마는 타입 힌트에서 자동 추론
 8@tool
 9def add(a: int, b: int) -> int:
10    """두 정수를 더한다."""
11    return a + b
12
13
14# 단계 2: 이름과 설명을 데코레이터 인자로 명시
15# 도구 이름은 calculator, 함수 이름은 _internal_calc로 분리 가능
16@tool("calculator", description="수학 계산을 정확하게 수행한다. LLM은 큰 숫자 계산에 약하므로 계산이 필요하면 이 도구를 쓴다.")
17def _internal_calc(expression: str) -> str:
18    return str(eval(expression, {"__builtins__": {}}, {}))
19
20
21# 단계 3: Pydantic으로 인자 스키마를 강제
22# 타입 힌트만으로 부족할 때 (검증 규칙, 기본값, 인자별 설명)
23class SearchEmployeeInput(BaseModel):
24    name: str = Field(
25        description="직원의 정확한 한국어 이름 (부분 일치 불가)",
26        min_length=2, max_length=20,   # 너무 짧거나 긴 값을 실행 전에 차단
27    )
28
29
30@tool(args_schema=SearchEmployeeInput)
31def search_employee(name: str) -> str:
32    """삼성SDS 임직원 디렉터리에서 이름으로 직원 정보를 조회한다.
33
34    직원의 소속 팀, 직무, 이메일이 필요할 때 사용한다.
35    사내 규정이나 정책 질문에는 사용하지 않는다.
36    없는 이름이면 "찾을 수 없습니다"를 돌려준다.
37    """
38    # 교육용 더미 데이터입니다
39    fake_db = {
40        "김하늘": "김하늘 / 클라우드운영팀 / 팀장 / [email protected]",
41        "박도윤": "박도윤 / 물류플랫폼팀 / 백엔드 / [email protected]",
42    }
43    return fake_db.get(name, f"'{name}' 직원을 찾을 수 없습니다.")
44
45
46if __name__ == "__main__":
47    # 도구는 함수가 아니라 객체이므로 .invoke()로 호출합니다
48    print(add.invoke({"a": 3, "b": 5}))
49    print(_internal_calc.invoke({"expression": "15 * 25"}))
50    print(search_employee.invoke({"name": "김하늘"}))
51
52    # Pydantic 검증: 잘못된 인자는 도구 실행 전에 차단됩니다
53    try:
54        search_employee.invoke({"name": "김"})   # min_length=2 위반
55    except Exception as e:
56        print(f"검증 에러 (예상된 동작): {type(e).__name__}")
57
58    print("=== 도구 메타데이터 (LLM이 실제로 보는 것) ===")
59    for t in [add, _internal_calc, search_employee]:
60        print(f"이름: {t.name}")
61        print(f"설명: {t.description}")
62        print(f"스키마: {t.args_schema.model_json_schema() if t.args_schema else None}")
63        print()

 1# tool_error.py - 실패를 메시지로 돌려주고 Agent가 복구하게 만들기
 2from langchain_core.tools import tool
 3
 4# 교육용 더미 데이터입니다!
 5EMPLOYEES = [
 6    {"name": "김하늘", "team": "클라우드운영팀", "email": "[email protected]"},
 7    {"name": "박도윤", "team": "물류플랫폼팀", "email": "[email protected]"},
 8    {"name": "이서준", "team": "클라우드운영팀", "email": "[email protected]"},
 9]
10
11
12@tool
13def search_employee(name: str) -> str:
14    """삼성SDS 임직원 디렉터리에서 이름으로 직원 정보를 조회한다.
15
16    Args:
17        name: 직원의 정확한 한국어 이름 (최소 2자)
18
19    Returns:
20        직원 정보 문자열, 또는 다음 행동을 안내하는 에러 메시지
21    """
22    try:
23        # 1) 입력 검증 실패 -> raise 대신 안내 메시지 반환
24        if not name or len(name) < 2:
25            return "에러: 이름이 너무 짧습니다. 최소 2자 이상의 정확한 한국어 이름으로 다시 호출하세요."
26
27        # 2) 정상 검색
28        for emp in EMPLOYEES:
29            if emp["name"] == name:
30                return f"{emp['name']} / {emp['team']} / {emp['email']}"
31
32        # 3) 결과 없음 -> 후보를 함께 알려주면 LLM이 다음 판단을 하기 쉬움
33        candidates = ", ".join(e["name"] for e in EMPLOYEES)
34        return f"'{name}' 직원을 찾을 수 없습니다. 등록된 직원: {candidates}"
35
36    except Exception as e:
37        # 4) 예상 못한 예외도 그래프를 죽이지 않고 메시지로 변환
38        return f"에러: 조회 중 문제가 발생했습니다 ({type(e).__name__}). 잠시 후 다시 시도하세요."
39
40
41if __name__ == "__main__":
42    print(search_employee.invoke({"name": "김"}))       # 너무 짧음
43    print(search_employee.invoke({"name": "홍길동"}))   # 없는 직원
44    print(search_employee.invoke({"name": "김하늘"}))   # 정상

 1# toolnode_parallel.py - ToolNode + 병렬 호출 + 에러 처리 확인
 2from typing import TypedDict, Annotated, Literal
 3from dotenv import load_dotenv
 4from langgraph.graph import StateGraph, START, END
 5from langgraph.graph.message import add_messages
 6from langgraph.prebuilt import ToolNode
 7from langchain_aws import ChatBedrockConverse
 8from langchain_core.tools import tool
 9from langchain_core.messages import HumanMessage, SystemMessage
10
11load_dotenv()
12
13
14def get_text(message):
15    """ChatBedrockConverse는 content를 블록 리스트로 주기도 하므로 텍스트만 모아 반환합니다."""
16    content = message.content
17    if isinstance(content, list):
18        return "".join(block.get("text", "") for block in content if isinstance(block, dict))
19    return content
20
21
22@tool
23def get_temperature(city: str) -> str:
24    """도시의 현재 온도를 조회한다 (교육용 더미 데이터)."""
25    data = {"서울": "3도", "부산": "8도", "제주": "11도"}
26    if city not in data:
27        raise ValueError(f"{city}는 지원하지 않는 도시입니다. 지원: {list(data)}")
28    return f"{city}: {data[city]}"
29
30
31@tool
32def divide(a: float, b: float) -> float:
33    """두 수를 나눈다."""
34    return a / b     # b가 0이면 ZeroDivisionError, ToolNode가 처리
35
36
37tools = [get_temperature, divide]
38
39
40class State(TypedDict):
41    messages: Annotated[list, add_messages]
42
43
44llm = ChatBedrockConverse(
45    model="us.anthropic.claude-sonnet-4-6",
46    region_name="us-east-1",
47).bind_tools(tools)
48
49
50def agent(state: State) -> dict:
51    msgs = [SystemMessage(content="도구를 사용해 답하세요.")] + state["messages"]
52    return {"messages": [llm.invoke(msgs)]}
53
54
55def should_continue(state: State) -> Literal["tools", "end"]:
56    return "tools" if state["messages"][-1].tool_calls else "end"
57
58
59builder = StateGraph(State)
60builder.add_node("agent", agent)
61builder.add_node("tools", ToolNode(tools, handle_tool_errors=True))
62builder.add_edge(START, "agent")
63builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
64builder.add_edge("tools", "agent")
65graph = builder.compile()
66
67
68if __name__ == "__main__":
69    # 1) 병렬 호출: 한 질문에 도구 두 개가 한 턴에 호출됨
70    print("=== 병렬 호출 ===")
71    result = graph.invoke({"messages": [HumanMessage(content="서울과 부산의 온도를 동시에 알려주세요.")]})
72    for m in result["messages"]:
73        if getattr(m, "tool_calls", None):
74            for tc in m.tool_calls:
75                print(f"  도구 호출 요청: {tc['name']}({tc['args']})")
76        elif m.type == "tool":
77            print(f"  도구 결과: {get_text(m)}")
78    print(f"최종 답변: {get_text(result['messages'][-1])}")
79
80    # 2) 에러 처리: 0으로 나누기 -> ToolMessage로 변환되어 Agent가 복구
81    print("\n=== 에러 처리 (handle_tool_errors=True) ===")
82    result = graph.invoke({"messages": [HumanMessage(content="10 나누기 0은?")]})
83    print(f"최종 답변: {get_text(result['messages'][-1])}")

 1# trace.py - messages 리스트를 순회하며 실행 과정을 표 형태로 출력
 2import json
 3from langchain_core.messages import HumanMessage
 4from toolnode_parallel import graph   # 지난 실습부분에서 사용한 그래프를 그대로 사용
 5
 6
 7def get_text(message):
 8    """ChatBedrockConverse는 content를 블록 리스트로 주기도 하므로 텍스트만 모아 반환합니다."""
 9    content = message.content
10    if isinstance(content, list):
11        return "".join(block.get("text", "") for block in content if isinstance(block, dict))
12    return content
13
14
15def print_trace(result: dict):
16    """messages 리스트를 순회하며 실행 과정을 표 형태로 출력."""
17    print(f"{'순서':<4} {'타입':<10} {'내용'}")
18    print("-" * 70)
19    for i, m in enumerate(result["messages"]):
20        if m.type == "human":
21            print(f"{i:<4} {'사용자':<10} {get_text(m)}")
22        elif m.type == "ai" and getattr(m, "tool_calls", None):
23            for tc in m.tool_calls:
24                print(f"{i:<4} {'도구요청':<10} {tc['name']} args={json.dumps(tc['args'], ensure_ascii=False)}")
25        elif m.type == "tool":
26            print(f"{i:<4} {'도구결과':<10} [{m.name}] {get_text(m)[:60]}")
27        elif m.type == "ai":
28            print(f"{i:<4} {'최종답변':<10} {get_text(m)[:60]}")
29
30
31if __name__ == "__main__":
32    result = graph.invoke({"messages": [HumanMessage(content="서울 온도를 100으로 나눈 값은?")]})
33    print_trace(result)

 1# make_data.py - Day4 백데이터 3종을 만드는 스크립트 (교육용 더미 데이터)
 2"""실행하면 같은 폴더에 employees.json, assets.json, projects.json을 만듭니다.
 3
 4실행: python make_data.py
 5
 65, 6, 7일차가 이 데이터를 그대로 이어 쓰므로 값은 바꾸지 마세요.
 7"""
 8import json
 9import os
10
11BASE = os.path.dirname(os.path.abspath(__file__))
12
13EMPLOYEES = [
14    {"id": 1, "name": "김하늘", "team": "클라우드운영팀", "role": "팀장", "email": "[email protected]"},
15    {"id": 2, "name": "박도윤", "team": "물류플랫폼팀", "role": "백엔드", "email": "[email protected]"},
16    {"id": 3, "name": "이서준", "team": "클라우드운영팀", "role": "인프라 엔지니어", "email": "[email protected]"},
17    {"id": 4, "name": "최민준", "team": "물류플랫폼팀", "role": "프론트엔드", "email": "[email protected]"},
18    {"id": 5, "name": "정하은", "team": "클라우드운영팀", "role": "데이터 분석가", "email": "[email protected]"},
19]
20
21# 노트북 5대, 모니터 2대, GPU 서버 1대로 총 8건입니다. GPU 서버는 6일차 MCP 통합 실습이 조회하므로 빼지 마세요.
22# 5-2의 총액 계산(노트북 5대 x 200만원 = 1,000만원)이 이 구성에 맞춰져 있으니 건수를 바꾸지 마세요.
23ASSETS = [
24    {"asset_id": "A-1001", "type": "노트북", "model": "갤럭시북4 프로", "owner": "김하늘", "status": "사용중"},
25    {"asset_id": "A-1002", "type": "노트북", "model": "갤럭시북4 프로", "owner": "박도윤", "status": "사용중"},
26    {"asset_id": "A-1003", "type": "노트북", "model": "갤럭시북4", "owner": "이서준", "status": "사용중"},
27    {"asset_id": "A-1004", "type": "노트북", "model": "갤럭시북4", "owner": None, "status": "재고"},
28    {"asset_id": "A-1005", "type": "노트북", "model": "갤럭시북4 프로", "owner": "최민준", "status": "사용중"},
29    {"asset_id": "A-1006", "type": "모니터", "model": "뷰피니티 S8", "owner": "김하늘", "status": "사용중"},
30    {"asset_id": "A-1007", "type": "모니터", "model": "뷰피니티 S8", "owner": "정하은", "status": "사용중"},
31    {"asset_id": "A-1008", "type": "GPU 서버", "model": "NVIDIA A100 80GB", "owner": "이서준", "status": "사용중"},
32]
33
34PROJECTS = [
35    {"project_id": "P-2601", "name": "차세대 물류 플랫폼 구축", "team": "물류플랫폼팀", "status": "진행중", "member_count": 12},
36    {"project_id": "P-2602", "name": "클라우드 비용 최적화", "team": "클라우드운영팀", "status": "진행중", "member_count": 7},
37    {"project_id": "P-2603", "name": "물류 관제 대시보드 고도화", "team": "물류플랫폼팀", "status": "진행중", "member_count": 5},
38    {"project_id": "P-2604", "name": "사내 AI 어시스턴트 도입", "team": "클라우드운영팀", "status": "진행중", "member_count": 4},
39]
40
41
42def write_json(filename: str, rows: list) -> str:
43    """한글이 깨지지 않게 UTF-8과 ensure_ascii=False로 씁니다."""
44    path = os.path.join(BASE, filename)
45    with open(path, "w", encoding="utf-8") as f:
46        json.dump(rows, f, ensure_ascii=False, indent=2)
47    return path
48
49
50if __name__ == "__main__":
51    for filename, rows in [("employees.json", EMPLOYEES),
52                           ("assets.json", ASSETS),
53                           ("projects.json", PROJECTS)]:
54        write_json(filename, rows)
55        print(f"생성: {filename} ({len(rows)}건)")
56
57    laptops = [a for a in ASSETS if a["type"] == "노트북"]
58    total = len(laptops) * 2000000
59    print(f"확인: 자산 {len(ASSETS)}건 중 노트북 {len(laptops)}대, 한 대당 200만원이면 총액 {total:,}원")

 1[
 2  {
 3    "id": 1,
 4    "name": "김하늘",
 5    "team": "클라우드운영팀",
 6    "role": "팀장",
 7    "email": "[email protected]"
 8  },
 9  {
10    "id": 2,
11    "name": "박도윤",
12    "team": "물류플랫폼팀",
13    "role": "백엔드",
14    "email": "[email protected]"
15  },
16  {
17    "id": 3,
18    "name": "이서준",
19    "team": "클라우드운영팀",
20    "role": "인프라 엔지니어",
21    "email": "[email protected]"
22  },
23  {
24    "id": 4,
25    "name": "최민준",
26    "team": "물류플랫폼팀",
27    "role": "프론트엔드",
28    "email": "[email protected]"
29  },
30  {
31    "id": 5,
32    "name": "정하은",
33    "team": "클라우드운영팀",
34    "role": "데이터 분석가",
35    "email": "[email protected]"
36  }
37]

 1[
 2  {
 3    "asset_id": "A-1001",
 4    "type": "노트북",
 5    "model": "갤럭시북4 프로",
 6    "owner": "김하늘",
 7    "status": "사용중"
 8  },
 9  {
10    "asset_id": "A-1002",
11    "type": "노트북",
12    "model": "갤럭시북4 프로",
13    "owner": "박도윤",
14    "status": "사용중"
15  },
16  {
17    "asset_id": "A-1003",
18    "type": "노트북",
19    "model": "갤럭시북4",
20    "owner": "이서준",
21    "status": "사용중"
22  },
23  {
24    "asset_id": "A-1004",
25    "type": "노트북",
26    "model": "갤럭시북4",
27    "owner": null,
28    "status": "재고"
29  },
30  {
31    "asset_id": "A-1005",
32    "type": "노트북",
33    "model": "갤럭시북4 프로",
34    "owner": "최민준",
35    "status": "사용중"
36  },
37  {
38    "asset_id": "A-1006",
39    "type": "모니터",
40    "model": "뷰피니티 S8",
41    "owner": "김하늘",
42    "status": "사용중"
43  },
44  {
45    "asset_id": "A-1007",
46    "type": "모니터",
47    "model": "뷰피니티 S8",
48    "owner": "정하은",
49    "status": "사용중"
50  },
51  {
52    "asset_id": "A-1008",
53    "type": "GPU 서버",
54    "model": "NVIDIA A100 80GB",
55    "owner": "이서준",
56    "status": "사용중"
57  }
58]

 1[
 2  {
 3    "project_id": "P-2601",
 4    "name": "차세대 물류 플랫폼 구축",
 5    "team": "물류플랫폼팀",
 6    "status": "진행중",
 7    "member_count": 12
 8  },
 9  {
10    "project_id": "P-2602",
11    "name": "클라우드 비용 최적화",
12    "team": "클라우드운영팀",
13    "status": "진행중",
14    "member_count": 7
15  },
16  {
17    "project_id": "P-2603",
18    "name": "물류 관제 대시보드 고도화",
19    "team": "물류플랫폼팀",
20    "status": "진행중",
21    "member_count": 5
22  },
23  {
24    "project_id": "P-2604",
25    "name": "사내 AI 어시스턴트 도입",
26    "team": "클라우드운영팀",
27    "status": "진행중",
28    "member_count": 4
29  }
30]

 1# mcp_server.py - 삼성SDS 교육용 더미 사내 데이터 MCP 서버 (fastmcp)
 2"""노출하는 도구 4개:
 3  - search_employee(name)      임직원 개인 조회
 4  - list_team_members(team)    팀별 임직원 목록
 5  - get_asset(owner)           자산 조회
 6  - list_projects(team)        프로젝트 목록
 7
 8실행: python mcp_server.py
 9(stdio 모드로 대기합니다.)
10"""
11import json
12import os
13from fastmcp import FastMCP
14
15mcp = FastMCP("sds-company-data")
16
17# make_data.py가 만든 JSON 3개를 코드와 같은 폴더에서 로드합니다 (교육용 더미 데이터)
18BASE = os.path.dirname(os.path.abspath(__file__))
19EMPLOYEES = json.load(open(os.path.join(BASE, "employees.json"), encoding="utf-8"))
20ASSETS = json.load(open(os.path.join(BASE, "assets.json"), encoding="utf-8"))
21PROJECTS = json.load(open(os.path.join(BASE, "projects.json"), encoding="utf-8"))
22
23
24@mcp.tool()
25def search_employee(name: str) -> str:
26    """삼성SDS 임직원 디렉터리에서 이름으로 직원 정보를 조회한다.
27
28    직원의 소속 팀, 직무, 이메일이 필요할 때 사용한다.
29    팀 전체 명단이나 인원수에는 사용하지 않는다 (list_team_members 사용).
30    자산이나 프로젝트 질문에도 사용하지 않는다 (get_asset, list_projects 사용).
31
32    Args:
33        name: 직원의 정확한 한국어 이름
34    """
35    for emp in EMPLOYEES:
36        if emp["name"] == name:
37            return json.dumps(emp, ensure_ascii=False)
38    names = ", ".join(e["name"] for e in EMPLOYEES)
39    return f"'{name}' 직원을 찾을 수 없습니다. 등록된 직원: {names}"
40
41
42@mcp.tool()
43def list_team_members(team: str) -> str:
44    """팀 이름으로 소속 임직원 목록을 조회한다.
45
46    특정 팀의 인원수나 명단이 필요할 때 사용한다.
47    개별 직원의 상세 정보는 search_employee를 사용한다.
48
49    Args:
50        team: 팀 이름 (예: '클라우드운영팀', '물류플랫폼팀')
51    """
52    rows = [e for e in EMPLOYEES if e["team"] == team]
53    if not rows:
54        teams = sorted({e["team"] for e in EMPLOYEES})
55        return f"'{team}' 팀을 찾을 수 없습니다. 등록된 팀: {teams}"
56    return json.dumps({"count": len(rows), "members": rows}, ensure_ascii=False, indent=2)
57
58
59@mcp.tool()
60def get_asset(owner: str = "") -> str:
61    """사내 자산(노트북, 모니터 등)을 조회한다.
62
63    Args:
64        owner: 사용자 이름으로 필터. 빈 문자열이면 전체 자산 반환
65    """
66    rows = [a for a in ASSETS if not owner or a["owner"] == owner]
67    if not rows:
68        return f"'{owner}' 사용자의 자산이 없습니다."
69    return json.dumps({"count": len(rows), "assets": rows}, ensure_ascii=False, indent=2)
70
71
72@mcp.tool()
73def list_projects(team: str = "") -> str:
74    """진행 중인 프로젝트 목록을 조회한다.
75
76    Args:
77        team: 팀 이름으로 필터 (예: '물류플랫폼팀'). 빈 문자열이면 전체
78    """
79    rows = [p for p in PROJECTS if not team or p["team"] == team]
80    if not rows:
81        return f"'{team}' 팀의 프로젝트가 없습니다."
82    return json.dumps(rows, ensure_ascii=False, indent=2)
83
84
85if __name__ == "__main__":
86    mcp.run()   # stdio 모드로 실행

 1# mcp_verify.py - 파이썬 MCP 클라이언트로 서버를 직접 검증
 2import asyncio
 3import os
 4import sys
 5from mcp import ClientSession, StdioServerParameters
 6from mcp.client.stdio import stdio_client
 7
 8SERVER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mcp_server.py")
 9
10
11async def main():
12    # 1) 서버를 자식 프로세스로 띄울 방법을 기술
13    #    sys.executable: 현재 파이썬 인터프리터 경로 (python/python3 문제 회피)
14    params = StdioServerParameters(command=sys.executable, args=[SERVER_PATH])
15
16    # 2) stdio로 연결하고 세션 시작
17    async with stdio_client(params) as (read, write):
18        async with ClientSession(read, write) as session:
19            await session.initialize()   # MCP 초기 핸드셰이크
20
21            # 3) 검증 1: 도구 목록 확인
22            tools = await session.list_tools()
23            print("=== 노출된 도구 목록 ===")
24            for t in tools.tools:
25                print(f"- {t.name}: {t.description.splitlines()[0]}")
26
27            # 4) 검증 2: 도구 호출
28            print("\n=== 도구 호출 검증 ===")
29            r1 = await session.call_tool("search_employee", {"name": "김하늘"})
30            print(f"search_employee -> {r1.content[0].text}")
31
32            r2 = await session.call_tool("list_team_members", {"team": "클라우드운영팀"})
33            print(f"list_team_members -> {r2.content[0].text[:80]}")
34
35            r3 = await session.call_tool("get_asset", {"owner": "김하늘"})
36            print(f"get_asset -> {r3.content[0].text}")
37
38            r4 = await session.call_tool("list_projects", {"team": "물류플랫폼팀"})
39            print(f"list_projects -> {r4.content[0].text}")
40
41            # 5) 검증 3: 에러 케이스도 확인
42            r5 = await session.call_tool("search_employee", {"name": "홍길동"})
43            print(f"없는 직원 -> {r5.content[0].text}")
44
45
46if __name__ == "__main__":
47    asyncio.run(main())

  1# mcp_agent.py - MCP 서버 + 일반 도구를 함께 쓰는 LangGraph 통합 Agent
  2import asyncio
  3import os
  4import sys
  5from dotenv import load_dotenv
  6from langchain_aws import ChatBedrockConverse
  7from langchain_core.tools import tool
  8from langchain_mcp_adapters.client import MultiServerMCPClient
  9from langgraph.prebuilt import create_react_agent
 10
 11load_dotenv()
 12
 13
 14def get_text(message):
 15    """ChatBedrockConverse는 content를 블록 리스트로 주기도 하므로 텍스트만 모아 반환합니다."""
 16    content = message.content
 17    if isinstance(content, list):
 18        return "".join(block.get("text", "") for block in content if isinstance(block, dict))
 19    return content
 20
 21
 22SERVER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mcp_server.py")
 23
 24SYSTEM_PROMPT = """당신은 삼성SDS 사내 어시스턴트입니다. 교육용 더미 데이터를 조회해 답합니다.
 25
 26[도구 사용 규칙]
 27- 임직원 개인 정보 -> search_employee
 28- 팀 인원수, 팀 명단 -> list_team_members
 29- 자산 현황 -> get_asset
 30- 프로젝트 현황 -> list_projects
 31- 수치 계산 -> calculator (암산하지 말고 반드시 도구 사용)
 32
 33도구에서 에러 메시지가 오면 그 지침을 따라 수정해서 다시 시도하세요.
 34도구 결과에 근거해서만 답하세요."""
 35
 36
 37# MCP 도구와 섞어 쓸 일반 도구
 38@tool
 39def calculator(expression: str) -> str:
 40    """수학 계산을 정확하게 수행한다. 예: '3 * 2000000'"""
 41    try:
 42        return str(eval(expression, {"__builtins__": {}}, {}))
 43    except Exception as e:
 44        return f"에러: 계산식이 잘못되었습니다 ({e}). 올바른 수식으로 다시 호출하세요."
 45
 46
 47async def main():
 48    # 1) MCP 클라이언트 설정: 서버 이름 -> 실행 방법
 49    client = MultiServerMCPClient({
 50        "sds-company-data": {
 51            "command": sys.executable,
 52            "args": [SERVER_PATH],
 53            "transport": "stdio",
 54        },
 55        # 서버를 더 붙이고 싶으면 여기에 항목 추가
 56    })
 57
 58    # 2) MCP 도구를 LangChain Tool로 변환해 가져오기 (await 필수)
 59    mcp_tools = await client.get_tools()
 60    print(f"MCP 도구 {len(mcp_tools)}개: {[t.name for t in mcp_tools]}")
 61
 62    # 3) 일반 도구와 합치기: 여기부터는 MCP 여부 구분이 사라집니다
 63    tools = mcp_tools + [calculator]
 64
 65    # 4) 2-2에서 손으로 조립한 ReAct 구조를 한 줄로
 66    llm = ChatBedrockConverse(
 67        model="us.anthropic.claude-sonnet-4-6",
 68        region_name="us-east-1",
 69    )
 70    agent = create_react_agent(llm, tools, prompt=SYSTEM_PROMPT)
 71
 72   
 73    # 5) 단일 도구 질문과 복합 질문을 섞어 검증
 74    queries = [
 75        # 단일 MCP 도구
 76        "박도윤 님의 이메일 주소를 알려주세요.",
 77        # 팀 단위 조회 (list_team_members)
 78        "클라우드운영팀 인원이 몇 명인지 알려주세요.",
 79        # MCP 도구 두 개를 이어야 하는 질문
 80        "이서준 님이 소속된 팀에서 진행 중인 프로젝트를 알려주세요.",
 81        # MCP 조회 + 일반 계산 도구를 엮는 복합 질문
 82        "노트북 자산이 총 몇 대인지 세고 한 대당 200만원이면 총액이 얼마인지 계산해 주세요.",
 83    ]
 84
 85    for q in queries:
 86        print(f"\n{'=' * 60}\n질문: {q}\n{'=' * 60}")
 87        async for event in agent.astream(
 88            {"messages": [("user", q)]},
 89            stream_mode="updates",
 90        ):
 91            for node, update in event.items():
 92                for m in update.get("messages", []):
 93                    if getattr(m, "tool_calls", None):
 94                        for tc in m.tool_calls:
 95                            print(f"  도구 호출: {tc['name']}({tc['args']})")
 96                    elif m.type == "tool":
 97                        print(f"  도구 결과: [{m.name}] {get_text(m)[:80]}")
 98                    elif m.content:
 99                        print(f"  답변: {get_text(m)[:200]}")
100
101
102if __name__ == "__main__":
103    asyncio.run(main())