AI Agent Function Calling 실전 구현: Tool Use 패턴·오류 처리·보안 설계와 프로덕션 배포 전략

from openai import OpenAI
import json

client = OpenAI()

# 도구 정의 - Strict Mode 적용
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_products",
            "description": "상품 카탈로그에서 조건에 맞는 상품을 검색합니다",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "검색 키워드"
                    },
                    "category": {
                        "type": ["string", "null"],
                        "description": "상품 카테고리 (electronics, clothing, food 등)",
                        "enum": ["electronics", "clothing", "food", None]
                    },
                    "max_price": {
                        "type": ["number", "null"],
                        "description": "최대 가격 (원 단위)"
                    },
                    "in_stock": {
                        "type": "boolean",
                        "description": "재고 있는 상품만 검색할지 여부"
                    }
                },
                "required": ["query", "category", "max_price", "in_stock"],
                "additionalProperties": False
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": "주문 번호로 배송 상태를 조회합니다",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "주문 번호 (예: ORD-20260312-001)"
                    }
                },
                "required": ["order_id"],
                "additionalProperties": False
            }
        }
    }
]


def execute_tool(name: str, arguments: dict) -> str:
    """도구 실행 디스패처"""
    if name == "search_products":
        # 실제로는 DB 또는 검색 엔진 호출
        return json.dumps({
            "results": [
                {"name": "무선 이어폰", "price": 89000, "stock": True},
                {"name": "블루투스 스피커", "price": 45000, "stock": True}
            ],
            "total": 2
        }, ensure_ascii=False)
    elif name == "get_order_status":
        return json.dumps({
            "order_id": arguments["order_id"],
            "status": "배송중",
            "estimated_delivery": "2026-03-14"
        }, ensure_ascii=False)
    return json.dumps({"error": "Unknown tool"})


def chat_with_tools(user_message: str) -> str:
    """도구를 사용하는 대화 루프"""
    messages = [
        {"role": "system", "content": "당신은 쇼핑몰 고객지원 AI입니다."},
        {"role": "user", "content": user_message}
    ]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )

    assistant_message = response.choices[0].message
    messages.append(assistant_message)

    # tool_calls가 있으면 실행 후 결과를 다시 전달
    if assistant_message.tool_calls:
        for tool_call in assistant_message.tool_calls:
            result = execute_tool(
                tool_call.function.name,
                json.loads(tool_call.function.arguments)
            )
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

        # 도구 결과를 포함하여 최종 응답 생성
        final_response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools
        )
        return final_response.choices[0].message.content

    return assistant_message.content


# 실행
result = chat_with_tools("5만원 이하 전자제품 검색해줘")
print(result)

import anthropic
import json

client = anthropic.Anthropic()

# Claude용 도구 정의
tools = [
    {
        "name": "search_database",
        "description": "고객 데이터베이스에서 정보를 검색합니다. 이름, 이메일, 주문 내역 등으로 검색 가능합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query_type": {
                    "type": "string",
                    "enum": ["customer", "order", "product"],
                    "description": "검색 대상 유형"
                },
                "search_term": {
                    "type": "string",
                    "description": "검색어"
                },
                "limit": {
                    "type": "integer",
                    "description": "최대 결과 수",
                    "default": 10
                }
            },
            "required": ["query_type", "search_term"]
        }
    },
    {
        "name": "send_notification",
        "description": "고객에게 이메일 또는 SMS 알림을 전송합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "recipient_id": {
                    "type": "string",
                    "description": "수신자 고객 ID"
                },
                "channel": {
                    "type": "string",
                    "enum": ["email", "sms"],
                    "description": "알림 채널"
                },
                "message": {
                    "type": "string",
                    "description": "알림 메시지 내용"
                }
            },
            "required": ["recipient_id", "channel", "message"]
        }
    }
]


def execute_claude_tool(name: str, tool_input: dict) -> str:
    """Claude 도구 실행"""
    if name == "search_database":
        return json.dumps({
            "results": [
                {"id": "C001", "name": "김영주", "email": "yj@example.com"}
            ]
        }, ensure_ascii=False)
    elif name == "send_notification":
        return json.dumps({
            "status": "sent",
            "message_id": "MSG-20260312-001"
        })
    return json.dumps({"error": "Unknown tool"})


def chat_with_claude_tools(user_message: str) -> str:
    """Claude Tool Use 대화 루프"""
    messages = [{"role": "user", "content": user_message}]

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        system="당신은 고객 관리 시스템의 AI 어시스턴트입니다.",
        tools=tools,
        messages=messages
    )

    # 에이전트 루프: stop_reason이 tool_use인 동안 반복
    while response.stop_reason == "tool_use":
        # 응답에서 tool_use 블록 추출
        tool_use_blocks = [
            block for block in response.content
            if block.type == "tool_use"
        ]

        # assistant 메시지 추가
        messages.append({"role": "assistant", "content": response.content})

        # 각 도구 실행 후 결과 전달
        tool_results = []
        for block in tool_use_blocks:
            result = execute_claude_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result
            })

        messages.append({"role": "user", "content": tool_results})

        # 다음 응답 요청
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system="당신은 고객 관리 시스템의 AI 어시스턴트입니다.",
            tools=tools,
            messages=messages
        )

    # 최종 텍스트 응답 추출
    text_blocks = [b.text for b in response.content if b.type == "text"]
    return "\n".join(text_blocks)


result = chat_with_claude_tools("김영주 고객 정보 조회하고 이메일로 배송 완료 알림 보내줘")
print(result)

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing import Optional
import json


# Pydantic 기반 도구 정의
class SearchProductsInput(BaseModel):
    """상품 검색을 위한 입력 스키마"""
    query: str = Field(description="검색 키워드")
    category: Optional[str] = Field(
        default=None,
        description="상품 카테고리"
    )
    max_price: Optional[int] = Field(
        default=None,
        description="최대 가격 (원)"
    )


class CreateTicketInput(BaseModel):
    """고객 문의 티켓 생성을 위한 입력 스키마"""
    title: str = Field(description="티켓 제목")
    description: str = Field(description="문의 내용")
    priority: str = Field(
        default="medium",
        description="우선순위: low, medium, high, urgent"
    )


@tool(args_schema=SearchProductsInput)
def search_products(query: str, category: Optional[str] = None,
                    max_price: Optional[int] = None) -> str:
    """상품 카탈로그에서 조건에 맞는 상품을 검색합니다."""
    results = [
        {"name": "MacBook Pro 14", "price": 2490000, "category": "electronics"},
        {"name": "AirPods Pro", "price": 359000, "category": "electronics"}
    ]
    if max_price:
        results = [r for r in results if r["price"] <= max_price]
    return json.dumps(results, ensure_ascii=False)


@tool(args_schema=CreateTicketInput)
def create_ticket(title: str, description: str,
                  priority: str = "medium") -> str:
    """고객 문의 티켓을 생성합니다."""
    return json.dumps({
        "ticket_id": "TKT-20260312-042",
        "status": "created",
        "priority": priority
    })


# 동일한 도구를 다른 LLM에 바인딩
tools_list = [search_products, create_ticket]

# OpenAI 모델 + 도구
openai_llm = ChatOpenAI(model="gpt-4o").bind_tools(tools_list)

# Anthropic 모델 + 동일한 도구
anthropic_llm = ChatAnthropic(
    model="claude-sonnet-4-20250514"
).bind_tools(tools_list)

# 호출 - 어떤 모델이든 동일한 인터페이스
response = openai_llm.invoke("50만원 이하 전자제품 찾아줘")
print(response.tool_calls)
# [{"name": "search_products", "args": {"query": "전자제품", "max_price": 500000}, "id": "..."}]

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# ReAct 에이전트 생성
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_react_agent(llm, tools=[search_products, create_ticket])

# 에이전트 실행 - 도구 호출과 추론을 자동으로 반복
result = agent.invoke({
    "messages": [
        {"role": "user", "content": "50만원 이하 전자제품 검색하고, 결과가 없으면 문의 티켓 만들어줘"}
    ]
})

for message in result["messages"]:
    print(f"[{message.type}] {message.content[:100] if message.content else ''}")

import asyncio
import random
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Optional


class CircuitState(Enum):
    CLOSED = "closed"        # 정상 동작
    OPEN = "open"            # 차단 상태
    HALF_OPEN = "half_open"  # 시험 요청 허용


@dataclass
class CircuitBreaker:
    """Circuit Breaker 패턴 구현"""
    failure_threshold: int = 5
    recovery_timeout: float = 30.0
    half_open_max_calls: int = 1

    state: CircuitState = field(default=CircuitState.CLOSED)
    failure_count: int = field(default=0)
    last_failure_time: float = field(default=0.0)
    half_open_calls: int = field(default=0)

    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
                return True
            return False
        # HALF_OPEN
        return self.half_open_calls < self.half_open_max_calls

    def record_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
        self.failure_count = 0

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN


def exponential_backoff_with_jitter(
    attempt: int,
    base_delay: float = 1.0,
    max_delay: float = 60.0
) -> float:
    """Exponential Backoff with Full Jitter"""
    delay = min(base_delay * (2 ** attempt), max_delay)
    return random.uniform(0, delay)


async def execute_tool_with_retry(
    tool_fn: Callable,
    arguments: dict,
    circuit_breaker: CircuitBreaker,
    max_retries: int = 3,
    timeout: float = 30.0
) -> dict:
    """재시도와 Circuit Breaker가 적용된 도구 실행"""
    if not circuit_breaker.can_execute():
        return {
            "error": "Circuit breaker is open",
            "retry_after": circuit_breaker.recovery_timeout
        }

    for attempt in range(max_retries + 1):
        try:
            result = await asyncio.wait_for(
                tool_fn(**arguments),
                timeout=timeout
            )
            circuit_breaker.record_success()
            return {"success": True, "data": result}

        except asyncio.TimeoutError:
            circuit_breaker.record_failure()
            if attempt < max_retries:
                delay = exponential_backoff_with_jitter(attempt)
                await asyncio.sleep(delay)
            else:
                return {"error": "Tool execution timed out after retries"}

        except Exception as e:
            circuit_breaker.record_failure()
            if attempt < max_retries:
                delay = exponential_backoff_with_jitter(attempt)
                await asyncio.sleep(delay)
            else:
                return {"error": f"Tool execution failed: {str(e)}"}

    return {"error": "Max retries exceeded"}

import re
import hashlib
import time
from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum


class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass
class ToolCallRequest:
    """도구 호출 요청 래퍼"""
    tool_name: str
    arguments: dict
    user_id: str
    session_id: str
    timestamp: float = field(default_factory=time.time)


@dataclass
class SecurityPolicy:
    """도구별 보안 정책"""
    allowed_tools: list = field(default_factory=list)
    max_calls_per_minute: int = 10
    require_confirmation: list = field(default_factory=list)
    blocked_patterns: list = field(default_factory=list)


class ToolCallSecurityMiddleware:
    """도구 호출 보안 검증 미들웨어"""

    # 프롬프트 인젝션 탐지 패턴
    INJECTION_PATTERNS = [
        r"ignore\s+(all\s+)?previous\s+instructions",
        r"system\s*prompt",
        r"you\s+are\s+now",
        r"forget\s+(all\s+)?your\s+instructions",
        r"override\s+(all\s+)?rules",
        r"act\s+as\s+(a\s+)?root",
        r"sudo\s+",
        r"admin\s+mode",
    ]

    # SQL 인젝션 탐지 패턴
    SQL_PATTERNS = [
        r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|UNION|ALTER)\b)",
        r"(--|;|/\*|\*/)",
        r"(\bOR\b\s+\b1\b\s*=\s*\b1\b)",
    ]

    # 경로 탐색 탐지 패턴
    PATH_TRAVERSAL_PATTERNS = [
        r"\.\./",
        r"\.\.\%2[fF]",
        r"/etc/(passwd|shadow)",
    ]

    def __init__(self, policy: SecurityPolicy):
        self.policy = policy
        self.call_history: dict = {}

    def validate(self, request: ToolCallRequest) -> dict:
        """모든 보안 검증을 순차적으로 수행"""
        checks = [
            self._check_allowlist,
            self._check_rate_limit,
            self._check_injection,
            self._check_sql_injection,
            self._check_path_traversal,
            self._check_argument_length,
        ]

        for check in checks:
            result = check(request)
            if not result["passed"]:
                return {
                    "allowed": False,
                    "reason": result["reason"],
                    "risk_level": result.get("risk_level", RiskLevel.HIGH)
                }

        risk = self._calculate_risk_score(request)
        needs_confirm = request.tool_name in self.policy.require_confirmation

        return {
            "allowed": True,
            "risk_level": risk,
            "requires_confirmation": needs_confirm
        }

    def _check_allowlist(self, request: ToolCallRequest) -> dict:
        if request.tool_name not in self.policy.allowed_tools:
            return {
                "passed": False,
                "reason": f"Tool '{request.tool_name}' is not in the allowlist",
                "risk_level": RiskLevel.CRITICAL
            }
        return {"passed": True}

    def _check_rate_limit(self, request: ToolCallRequest) -> dict:
        key = f"{request.user_id}:{request.tool_name}"
        now = time.time()
        history = self.call_history.get(key, [])
        # 1분 이내의 호출만 유지
        recent = [t for t in history if now - t < 60]
        self.call_history[key] = recent

        if len(recent) >= self.policy.max_calls_per_minute:
            return {
                "passed": False,
                "reason": "Rate limit exceeded",
                "risk_level": RiskLevel.MEDIUM
            }
        self.call_history[key].append(now)
        return {"passed": True}

    def _check_injection(self, request: ToolCallRequest) -> dict:
        text = str(request.arguments).lower()
        for pattern in self.INJECTION_PATTERNS:
            if re.search(pattern, text, re.IGNORECASE):
                return {
                    "passed": False,
                    "reason": f"Prompt injection detected: {pattern}",
                    "risk_level": RiskLevel.CRITICAL
                }
        return {"passed": True}

    def _check_sql_injection(self, request: ToolCallRequest) -> dict:
        text = str(request.arguments)
        for pattern in self.SQL_PATTERNS:
            if re.search(pattern, text, re.IGNORECASE):
                return {
                    "passed": False,
                    "reason": "SQL injection pattern detected",
                    "risk_level": RiskLevel.CRITICAL
                }
        return {"passed": True}

    def _check_path_traversal(self, request: ToolCallRequest) -> dict:
        text = str(request.arguments)
        for pattern in self.PATH_TRAVERSAL_PATTERNS:
            if re.search(pattern, text):
                return {
                    "passed": False,
                    "reason": "Path traversal pattern detected",
                    "risk_level": RiskLevel.CRITICAL
                }
        return {"passed": True}

    def _check_argument_length(self, request: ToolCallRequest) -> dict:
        for key, value in request.arguments.items():
            if isinstance(value, str) and len(value) > 10000:
                return {
                    "passed": False,
                    "reason": f"Argument '{key}' exceeds max length",
                    "risk_level": RiskLevel.MEDIUM
                }
        return {"passed": True}

    def _calculate_risk_score(self, request: ToolCallRequest) -> RiskLevel:
        if request.tool_name in self.policy.require_confirmation:
            return RiskLevel.HIGH
        return RiskLevel.LOW


# 사용 예시
policy = SecurityPolicy(
    allowed_tools=["search_products", "get_order_status", "create_ticket"],
    max_calls_per_minute=10,
    require_confirmation=["create_ticket"],
    blocked_patterns=["delete", "drop", "truncate"]
)

middleware = ToolCallSecurityMiddleware(policy)

# 정상 요청
normal_request = ToolCallRequest(
    tool_name="search_products",
    arguments={"query": "노트북", "max_price": 2000000},
    user_id="user-001",
    session_id="sess-001"
)
print(middleware.validate(normal_request))
# {"allowed": True, "risk_level": RiskLevel.LOW, ...}

# 악의적 요청 - 허용되지 않은 도구
malicious_request = ToolCallRequest(
    tool_name="delete_all_users",
    arguments={"confirm": True},
    user_id="user-001",
    session_id="sess-001"
)
print(middleware.validate(malicious_request))
# {"allowed": False, "reason": "Tool 'delete_all_users' is not in the allowlist", ...}

# MCP 서버 구현 예시 (Python SDK)
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("ecommerce-tools")


@mcp.tool()
def search_products(query: str, max_price: int = None) -> str:
    """상품 카탈로그에서 조건에 맞는 상품을 검색합니다.

    Args:
        query: 검색 키워드
        max_price: 최대 가격 (원 단위, 선택)
    """
    # 실제 검색 로직
    import json
    results = [
        {"name": "MacBook Pro", "price": 2490000},
        {"name": "iPad Air", "price": 899000}
    ]
    if max_price:
        results = [r for r in results if r["price"] <= max_price]
    return json.dumps(results, ensure_ascii=False)


@mcp.tool()
def get_order_status(order_id: str) -> str:
    """주문 상태를 조회합니다.

    Args:
        order_id: 주문 번호
    """
    import json
    return json.dumps({
        "order_id": order_id,
        "status": "shipped",
        "tracking_number": "KR1234567890"
    })


@mcp.resource("products://catalog")
def get_product_catalog() -> str:
    """전체 상품 카탈로그를 리소스로 제공합니다."""
    import json
    return json.dumps({
        "categories": ["electronics", "clothing", "food"],
        "total_products": 15420
    })


# 서버 실행
if __name__ == "__main__":
    mcp.run(transport="streamable-http")

import time
import logging
from dataclasses import dataclass, field
from typing import Any, Optional

logger = logging.getLogger(__name__)


@dataclass
class AgentConfig:
    """에이전트 실행 설정"""
    max_iterations: int = 15
    max_execution_time: float = 120.0
    max_token_budget: int = 100000
    max_consecutive_same_tool: int = 3


@dataclass
class AgentState:
    """에이전트 실행 상태"""
    iteration: int = 0
    total_tokens: int = 0
    start_time: float = field(default_factory=time.time)
    tool_call_history: list = field(default_factory=list)
    last_tool_name: Optional[str] = None
    consecutive_same_tool: int = 0


class ProductionAgentLoop:
    """프로덕션 에이전트 루프"""

    def __init__(self, llm_client, tools: dict,
                 security_middleware, config: AgentConfig = None):
        self.llm = llm_client
        self.tools = tools
        self.security = security_middleware
        self.config = config or AgentConfig()

    async def run(self, messages: list, user_id: str,
                  session_id: str) -> dict:
        state = AgentState()

        while True:
            # 종료 조건 검사
            termination = self._check_termination(state)
            if termination:
                logger.warning(f"Agent terminated: {termination}")
                return {
                    "status": "terminated",
                    "reason": termination,
                    "messages": messages
                }

            # LLM 호출
            state.iteration += 1
            response = await self.llm.create(messages=messages)
            state.total_tokens += response.usage.total_tokens

            # 도구 호출이 없으면 최종 응답
            if not response.tool_calls:
                return {
                    "status": "completed",
                    "response": response.content,
                    "messages": messages,
                    "metrics": {
                        "iterations": state.iteration,
                        "total_tokens": state.total_tokens,
                        "execution_time": time.time() - state.start_time
                    }
                }

            # 도구 호출 처리
            messages.append(response.to_message())

            for tool_call in response.tool_calls:
                # 연속 동일 도구 호출 검사
                if tool_call.name == state.last_tool_name:
                    state.consecutive_same_tool += 1
                else:
                    state.consecutive_same_tool = 0
                state.last_tool_name = tool_call.name

                # 보안 검증
                request = ToolCallRequest(
                    tool_name=tool_call.name,
                    arguments=tool_call.arguments,
                    user_id=user_id,
                    session_id=session_id
                )
                validation = self.security.validate(request)

                if not validation["allowed"]:
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": f"Error: {validation['reason']}"
                    })
                    continue

                # 도구 실행
                try:
                    tool_fn = self.tools[tool_call.name]
                    result = await tool_fn(**tool_call.arguments)
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": str(result)
                    })
                    state.tool_call_history.append({
                        "tool": tool_call.name,
                        "timestamp": time.time(),
                        "success": True
                    })
                except Exception as e:
                    logger.error(f"Tool execution error: {e}")
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": f"Error executing tool: {str(e)}"
                    })
                    state.tool_call_history.append({
                        "tool": tool_call.name,
                        "timestamp": time.time(),
                        "success": False,
                        "error": str(e)
                    })

    def _check_termination(self, state: AgentState) -> Optional[str]:
        if state.iteration >= self.config.max_iterations:
            return f"Max iterations reached ({self.config.max_iterations})"
        elapsed = time.time() - state.start_time
        if elapsed >= self.config.max_execution_time:
            return f"Max execution time reached ({self.config.max_execution_time}s)"
        if state.total_tokens >= self.config.max_token_budget:
            return f"Token budget exhausted ({self.config.max_token_budget})"
        if state.consecutive_same_tool >= self.config.max_consecutive_same_tool:
            return f"Same tool called {self.config.max_consecutive_same_tool} times consecutively"
        return None

사용자 요청
    │
    ▼
┌──────────────────┐
│  Router Agent    │  ← 의도 분류 및 라우팅
└────────┬─────────┘
         │
    ┌────┼────┬────────┐
    ▼    ▼    ▼        ▼
┌──────┐┌──────┐┌──────┐┌──────┐
│검색  ││주문  ││CS    ││결제  │
│Agent ││Agent ││Agent ││Agent │
└──────┘└──────┘└──────┘└──────┘