MCP PII Tools

Instructions

MCP Tool: 텍스트에서 PII 탐지

Args:
    text (str): 분석할 텍스트
    
Returns:
    Dict[str, Any]: 탐지 결과

Implementation Reference

• mcp_pii_tools.py:528-541 (handler)

  The handler function for the 'mcp_detect_pii' MCP tool. Decorated with @mcp.tool() for automatic registration and execution. Delegates to the global detector's detect_pii method.

  @mcp.tool()
  def mcp_detect_pii(text: str) -> Dict[str, Any]:
      """
      MCP Tool: 텍스트에서 PII 탐지
      
      Args:
          text (str): 분석할 텍스트
          
      Returns:
          Dict[str, Any]: 탐지 결과
      """
      detector = get_detector()
      return detector.detect_pii(text)

• mcp_pii_tools.py:154-256 (helper)

  Core helper method in MCPPIIDetector class that performs the actual PII detection using langextract library, handles positions, confidence, and formats the output.

  def detect_pii(self, text: str) -> Dict[str, Any]:
      """
      텍스트에서 PII를 탐지 (MCP Tool용)
      
      Args:
          text (str): 분석할 텍스트
          
      Returns:
          Dict[str, Any]: MCP Tool 응답 형식
      """
      try:
          start_time = time.time()
          
          # Provider에 따른 langextract 호출
          if self.provider_type == "vllm":
              # vLLM Provider 사용
              result = lx.extract(
                  text_or_documents=text,
                  prompt_description=self.prompt,
                  examples=self.examples,
                  model=self.provider,  # 커스텀 Provider 인스턴스 사용
                  use_schema_constraints=False
              )
          else:
              # OpenAI Provider 사용 (기본)
              os.environ["OPENAI_BASE_URL"] = "https://api.openai.com/v1"
              result = lx.extract(
                  text_or_documents=text,
                  prompt_description=self.prompt,
                  examples=self.examples,
                  model_id=self.model_id,
                  api_key=self.api_key,
                  fence_output=True
              )
          
          # 결과를 PIIItem 리스트로 변환
          pii_items = []
          logger.info(f"탐지된 extraction 수: {len(result.extractions)}")
          
          for i, extraction in enumerate(result.extractions):
              logger.info(f"Extraction {i+1}: class='{extraction.extraction_class}', text='{extraction.extraction_text}'")
              
              # char_interval이 없으면 텍스트에서 직접 위치 찾기
              start_pos = 0
              end_pos = 0
              
              if extraction.char_interval:
                  start_pos = extraction.char_interval.start_pos
                  end_pos = extraction.char_interval.end_pos
                  logger.info(f"  char_interval 사용: {start_pos}-{end_pos}")
              else:
                  # 텍스트에서 직접 위치 찾기 (대소문자 구분 없이)
                  search_text = extraction.extraction_text
                  start_pos = text.find(search_text)
                  
                  # 대소문자 구분 없이 찾기
                  if start_pos == -1:
                      start_pos = text.lower().find(search_text.lower())
                  
                  if start_pos != -1:
                      end_pos = start_pos + len(search_text)
                      logger.info(f"  텍스트에서 직접 찾음: {start_pos}-{end_pos}")
                      
                      # 실제 찾은 텍스트와 원본이 일치하는지 확인
                      actual_found = text[start_pos:end_pos]
                      if actual_found != search_text:
                          logger.warning(f"  대소문자 차이: 찾은='{actual_found}', 원본='{search_text}'")
                  else:
                      start_pos = -1
                      end_pos = -1
                      logger.warning(f"  텍스트에서 찾을 수 없음: '{extraction.extraction_text}'")
              
              mapped_type = self._map_extraction_class(extraction.extraction_class)
              logger.info(f"  매핑된 타입: '{extraction.extraction_class}' -> '{mapped_type}'")
              
              pii_items.append(PIIItem(
                  type=mapped_type,
                  value=extraction.extraction_text,
                  confidence=0.9,  # langextract는 confidence를 제공하지 않으므로 기본값
                  start_pos=start_pos,
                  end_pos=end_pos
              ))
          
          processing_time = time.time() - start_time
          
          return {
              "success": True,
              "pii_items": [asdict(item) for item in pii_items],
              "count": len(pii_items),
              "processing_time": processing_time,
              "summary": self._get_pii_summary(pii_items)
          }
          
      except Exception as e:
          return {
              "success": False,
              "error": str(e),
              "pii_items": [],
              "count": 0,
              "processing_time": 0,
              "summary": {}
          }

• mcp_pii_tools.py:41-49 (schema)

  Dataclass schema defining the structure of individual PII detections returned by the tool.

  @dataclass
  class PIIItem:
      """PII 항목을 나타내는 데이터 클래스"""
      type: str           # PII 유형 (이름, 전화번호, 이메일 등)
      value: str          # 추출된 값
      confidence: float   # 신뢰도 (0.0-1.0)
      start_pos: int      # 텍스트 내 시작 위치
      end_pos: int        # 텍스트 내 끝 위치

• mcp_pii_tools.py:31-37 (helper)

  Helper function providing singleton access to the global MCPPIIDetector instance used by the tool.

  def get_detector():
      """전역 PII 탐지기 인스턴스를 반환합니다."""
      global _global_detector
      if _global_detector is None:
          _global_detector = MCPPIIDetector()
      return _global_detector

• mcp_pii_tools.py:841-854 (schema)

  Explicit JSON schema for the tool's input parameters (text: string), used for documentation and validation.

  "detect_pii": {
      "name": "detect_pii",
      "description": "텍스트에서 PII(개인정보)를 탐지합니다. 이름, 이메일, 전화번호, 여권번호, 주소 등을 찾습니다.",
      "parameters": {
          "type": "object",
          "properties": {
              "text": {
                  "type": "string",
                  "description": "분석할 텍스트"
              }
          },
          "required": ["text"]
      }
  },