mcp_tools.py•5.31 kB
import subprocess
import time
from mcp.server.fastmcp import FastMCP
from chatgpt_mcp.chatgpt_automation import ChatGPTAutomation, check_chatgpt_access
def is_conversation_complete() -> bool:
"""Check if ChatGPT conversation is complete using external AppleScript.
Returns:
True if conversation is complete, False if still in progress
"""
try:
automation = ChatGPTAutomation()
screen_data = automation.read_screen_content()
if screen_data.get("status") == "success":
indicators = screen_data.get("indicators", {})
# Simple check: only use conversationComplete indicator
result = indicators.get("conversationComplete", False)
print(f"[DEBUG] is_conversation_complete: {result}, indicators: {indicators}")
return result
else:
print(f"[DEBUG] Screen read failed: {screen_data}")
# If we can't read the screen, assume not complete for safety
return False
except Exception as e:
print(f"[DEBUG] Exception in is_conversation_complete: {e}")
# If any error occurs, assume not complete for safety
return False
def wait_for_response_completion(max_wait_time: int = 300, check_interval: float = 2) -> bool:
"""Wait for ChatGPT response to complete.
Args:
max_wait_time: Maximum time to wait in seconds
check_interval: How often to check for completion in seconds
Returns:
True if response completed within time limit, False if timed out
"""
start_time = time.time()
while time.time() - start_time < max_wait_time:
if is_conversation_complete():
return True
time.sleep(check_interval)
return False
def get_current_conversation_text() -> str:
"""Get the current conversation text from ChatGPT.
Returns:
Current conversation text (last 5 messages)
"""
try:
automation = ChatGPTAutomation()
last_messages = automation.get_last_messages(count=5)
if last_messages:
return last_messages
else:
return "No response received from ChatGPT."
except Exception as e:
return f"Error reading conversation: {str(e)}"
async def get_chatgpt_response() -> str:
"""Get the latest response from ChatGPT after sending a message.
Returns:
ChatGPT's latest response text
"""
try:
# Wait for response to complete
if wait_for_response_completion():
return get_current_conversation_text()
else:
return "Timeout: ChatGPT response did not complete within the time limit."
except Exception as e:
raise Exception(f"Failed to get response from ChatGPT: {str(e)}")
async def ask_chatgpt(prompt: str) -> str:
"""Send a prompt to ChatGPT and return the response.
Args:
prompt: The text to send to ChatGPT
Returns:
ChatGPT's response
"""
await check_chatgpt_access()
try:
# 프롬프트에서 개행 문자 제거 및 더블쿼츠를 싱글쿼츠로 변경
cleaned_prompt = prompt.replace('\n', ' ').replace('\r', ' ').replace('"', "'").strip()
# Activate ChatGPT and send message using keystroke
automation = ChatGPTAutomation()
automation.activate_chatgpt()
automation.send_message_with_keystroke(cleaned_prompt)
# Get the response
response = await get_chatgpt_response()
return response
except Exception as e:
raise Exception(f"Failed to send message to ChatGPT: {str(e)}")
async def new_chatgpt_chat() -> str:
"""Start a new chat conversation in ChatGPT.
Returns:
Success message or error
"""
await check_chatgpt_access()
try:
automation = ChatGPTAutomation()
result = automation.new_chat()
if isinstance(result, tuple):
success, method = result
if success:
return f"Successfully opened a new ChatGPT chat window using: {method}"
else:
return f"Failed to open a new chat window. Last tried method: {method}"
else:
# 이전 버전과의 호환성
if result:
return "Successfully opened a new ChatGPT chat window."
else:
return "Failed to open a new chat window. Please check if ChatGPT window is in the foreground."
except Exception as e:
raise Exception(f"Failed to create new chat: {str(e)}")
def setup_mcp_tools(mcp: FastMCP):
"""MCP 도구들을 설정"""
@mcp.tool()
async def ask_chatgpt_tool(prompt: str) -> str:
"""Send a prompt to ChatGPT and return the response."""
return await ask_chatgpt(prompt)
@mcp.tool()
async def get_chatgpt_response_tool() -> str:
"""Get the latest response from ChatGPT after sending a message."""
return await get_chatgpt_response()
@mcp.tool()
async def new_chatgpt_chat_tool() -> str:
"""Start a new chat conversation in ChatGPT."""
return await new_chatgpt_chat()