openai-tool2mcp

Instructions

Implementation Reference

• openai_tool2mcp/tools/web_search.py:9-116 (handler)

  WebSearchAdapter implements the handler logic for the 'web-search' tool, translating MCP requests (e.g., search_term to query) to OpenAI parameters and OpenAI responses back to MCPResponse with robust JSON/string handling.
  class WebSearchAdapter(ToolAdapter): """Adapter for OpenAI's web search tool""" @property def tool_id(self) -> str: """Get the MCP tool ID""" return "web-search" @property def openai_tool_type(self) -> str: """Get the OpenAI tool type""" return "web_search" @property def description(self) -> str: """Get the tool description""" return "Search the web for real-time information" async def translate_request(self, request: MCPRequest) -> dict: """ Translate MCP request to OpenAI parameters. Args: request (MCPRequest): MCP request Returns: dict: Dictionary of OpenAI parameters """ parameters = {} if "search_term" in request.parameters: parameters["query"] = request.parameters["search_term"] elif "parameters" in request.parameters: # 경우에 따라 parameters 필드 내에 직접 쿼리가 문자열로 전달될 수 있음 parameters["query"] = request.parameters["parameters"] # 향상된 로깅 추가 logger.info(f"[WEB SEARCH] Translated MCP request to tool parameters: {parameters}") return parameters async def translate_response(self, response: Any) -> MCPResponse: """ Translate OpenAI tool response to MCP response. Args: response (Any): OpenAI tool response (JSON string or dictionary) Returns: MCPResponse: MCP response """ # 향상된 디버그 로깅 추가 logger.debug(f"[WEB SEARCH] Translating response of type: {type(response)}") # 응답이 이미 JSON 문자열인지 확인 if isinstance(response, str): try: # 이미 JSON 문자열인 경우 파싱 logger.debug("[WEB SEARCH] Response is a string, attempting to parse as JSON") parsed_response = json.loads(response) # 응답 내용 확인 및 추출 logger.debug( f"[WEB SEARCH] Parsed JSON with keys: {list(parsed_response.keys()) if isinstance(parsed_response, dict) else 'not a dict'}" ) # 콘텐츠 추출 (문자열로 변환) if isinstance(parsed_response, dict) and "content" in parsed_response: content = str(parsed_response["content"]) logger.info(f"[WEB SEARCH] Extracted content from JSON response, length: {len(content)}") return MCPResponse(content=content, context={"raw_response": response}) else: # 콘텐츠 키가 없으면 전체 응답을 문자열로 변환 content_str = json.dumps(parsed_response) logger.info( f"[WEB SEARCH] No content key found, using full response as content, length: {len(content_str)}" ) return MCPResponse(content=content_str, context={"raw_response": response}) except json.JSONDecodeError: # JSON 파싱 실패 시 원본 문자열 사용 logger.warning("[WEB SEARCH] Failed to parse response as JSON, using raw string") return MCPResponse(content=response, context={"raw_response": response}) # 딕셔너리인 경우 elif isinstance(response, dict): logger.debug(f"[WEB SEARCH] Response is a dict with keys: {list(response.keys())}") # 콘텐츠 키가 있으면 해당 값 사용 if "content" in response: # 반드시 content를 문자열로 변환 content = str(response["content"]) logger.info(f"[WEB SEARCH] Extracted content from dict response, length: {len(content)}") return MCPResponse(content=content, context={"raw_response": json.dumps(response)}) # 콘텐츠 키가 없으면 전체 딕셔너리를 JSON 문자열로 변환 content_str = json.dumps(response) logger.info( f"[WEB SEARCH] No content key in dict, using full dict as JSON string, length: {len(content_str)}" ) return MCPResponse(content=content_str, context={"raw_response": content_str}) # 다른 모든 타입의 경우 else: # 안전하게 문자열로 변환 logger.warning(f"[WEB SEARCH] Unexpected response type: {type(response)}, converting to string") content_str = str(response) return MCPResponse(content=content_str, context={"raw_response": content_str})
• openai_tool2mcp/server.py:52-60 (registration)

  Instantiates WebSearchAdapter and maps it to 'web-search' tool_id in the server's tools_map, enabling its use in MCP tool registration.
  # Register default tool adapters adapters = [WebSearchAdapter(), CodeInterpreterAdapter(), BrowserAdapter(), FileManagerAdapter()] for adapter in adapters: # Only register if the tool is enabled if adapter.openai_tool_type in self.config.tools: tools_map[adapter.tool_id] = adapter return tools_map
• openai_tool2mcp/tools/registry.py:38-42 (registration)

  Registers the 'web-search' MCP tool in ToolRegistry, mapping it to OpenAI's 'web_search' tool with enable/disable logic.
  "web-search": { "openai_tool": OpenAIBuiltInTools.WEB_SEARCH.value, "enabled": OpenAIBuiltInTools.WEB_SEARCH.value in self.enabled_tools, "description": "Search the web for information", },
• openai_tool2mcp/server.py:62-114 (registration)

  Registers the 'web-search' tool with the MCP SDK using @mcp.tool(name=tool_id), with a handler that uses the WebSearchAdapter for translation and invokes OpenAI client.
  def _register_mcp_tools(self): """Register tools with the MCP SDK""" for tool_id, adapter in self.tools_map.items(): # Define a tool handler for each adapter # Create a closure to properly capture the values def create_tool_handler(tool_id=tool_id, adapter=adapter): @self.mcp.tool(name=tool_id, description=adapter.description) async def tool_handler(**parameters): """ MCP tool handler for OpenAI tools. """ # Create an MCP request from the parameters mcp_request = MCPRequest(parameters=parameters) # Translate the request parameters using the adapter translated_params = await adapter.translate_request(mcp_request) # Create an OpenAI tool request openai_request = mcp_to_openai.translate_request(mcp_request, tool_id) # Override the parameters with the adapter-specific ones openai_request.parameters = translated_params try: # Call OpenAI API to execute the tool openai_response = await self.openai_client.invoke_tool(openai_request) # Translate the OpenAI response to MCP format using the adapter if openai_response.tool_outputs: # Use the adapter to translate the tool-specific response mcp_response = await adapter.translate_response(openai_response.tool_outputs[0].output) # Add thread_id to context for state management if mcp_response.context is None: mcp_response.context = {} mcp_response.context["thread_id"] = openai_response.thread_id # Return the response content which will be used by MCP SDK return mcp_response.content else: # Fallback to generic translation mcp_response = openai_to_mcp.translate_response(openai_response) return mcp_response.content except Exception as e: logger.error(f"Error invoking tool {tool_id}: {e!s}") # Using custom exception class to fix TRY003 raise ToolInvocationError() from e return tool_handler # Create and register the tool handler create_tool_handler()