OpenAI Assistant MCP Server

import logging from openai import AsyncOpenAI logger = logging.getLogger(__name__) class LLMConnector: def __init__(self, openai_api_key: str): self.client = AsyncOpenAI(api_key=openai_api_key) async def ask_openai(self, query: str, model: str = "gpt-4", temperature: float = 0.7, max_tokens: int = 500) -> str: try: response = await self.client.chat.completions.create( messages=[ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": query} ], model=model, temperature=temperature, max_tokens=max_tokens ) return response.choices[0].message.content except Exception as e: logger.error(f"Failed to query OpenAI: {str(e)}") raise async def list_assistants(self): try: response = await self.client.beta.assistants.list() return response.data except Exception as e: logger.error(f"Failed to list assistants: {str(e)}") raise async def retrieve_assistant(self, assistant_id: str): try: response = await self.client.beta.assistants.retrieve(assistant_id) return response except Exception as e: logger.error(f"Failed to retrieve assistant {assistant_id}: {str(e)}") raise async def create_assistant(self, name: str, instructions: str, model: str, tools: list = None, file_ids: list = None, temperature: float = 0.7): try: assistant = await self.client.beta.assistants.create( name=name, instructions=instructions, model=model, tools=tools or [{"type": "code_interpreter"}], # Default tool tool_resources={'file_search': {'vector_store_ids': file_ids}} if file_ids else None, temperature=temperature ) return assistant except Exception as e: logger.error(f"Failed to create assistant: {str(e)}") raise async def update_assistant(self, assistant_id: str, **kwargs): try: # Filter out None values to avoid overwriting existing settings with defaults update_data = {k: v for k, v in kwargs.items() if v is not None} # Handle file_ids separately to fit the tool_resources structure if 'file_ids' in update_data: file_ids = update_data.pop('file_ids') if file_ids: update_data['tool_resources'] = {'file_search': {'vector_store_ids': file_ids}} else: # To remove files update_data['tool_resources'] = {} if not update_data: raise ValueError("No update data provided") response = await self.client.beta.assistants.update(assistant_id, **update_data) return response except Exception as e: logger.error(f"Failed to update assistant {assistant_id}: {str(e)}") raise async def delete_assistant(self, assistant_id: str): try: response = await self.client.beta.assistants.delete(assistant_id) return response except Exception as e: logger.error(f"Failed to delete assistant {assistant_id}: {str(e)}") raise async def upload_file(self, file_path: str): try: with open(file_path, "rb") as file: response = await self.client.files.create(file=file, purpose="assistants") return response except Exception as e: logger.error(f"Failed to upload file {file_path}: {str(e)}") raise async def list_files(self): try: response = await self.client.files.list(purpose="assistants") return response.data except Exception as e: logger.error(f"Failed to list files: {str(e)}") raise async def delete_file(self, file_id: str): try: response = await self.client.files.delete(file_id) return response except Exception as e: logger.error(f"Failed to delete file {file_id}: {str(e)}") raise