OpenAI Vector Store MCP Server

import json import uuid import time import threading from sseclient import SSEClient from openai import OpenAI import os from dotenv import load_dotenv # Load environment variables load_dotenv() OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") VECTOR_STORE_ID = os.getenv("VECTOR_STORE_ID") SERVER_URL = "http://localhost:8000/sse" client = OpenAI(api_key=OPENAI_API_KEY) # ---- Utility: vector store file list ---- def list_vector_store_files(): print("\n📁 Loading vector store files...\n") files = client.vector_stores.files.list(vector_store_id=VECTOR_STORE_ID) items = files.data if not items: print("❌ No files found") return None for f in items: print(f"- {f.id} | {f.filename}") return items[0].id # return first file id # ---- Send event to MCP server ---- def send_mcp_message(sse_connection, message_type, payload): """ FastMCP는 SSE 송신을 위해 connection 객체에 send_event 사용 단, SSEClient는 read-only 이므로 별도 thread로 서버에 이벤트 POST FastMCP에서는 /sse endpoint로 이벤트 POST 가능 """ import requests data = { "type": message_type, **payload, "message_id": str(uuid.uuid4()) } print(f"\n➡️ Sending MCP message ({message_type}):") print(json.dumps(data, indent=2, ensure_ascii=False)) r = requests.post( SERVER_URL, json=data, headers={"Content-Type": "application/json"} ) print("POST status:", r.status_code) if r.status_code >= 400: print("Server returned error:") print(r.text) # ---- Main logic ---- def run_test(): print("\n🔌 Connecting to MCP server SSE stream...\n") messages = SSEClient(SERVER_URL) # Start receiving in another thread def listen(): for msg in messages: if not msg.data.strip(): continue print("\n📥 MCP EVENT RECEIVED:") try: parsed = json.loads(msg.data) print(json.dumps(parsed, indent=2, ensure_ascii=False)) except: print(msg.data) listener = threading.Thread(target=listen, daemon=True) listener.start() time.sleep(1) # Search test send_mcp_message( messages, "call_tool", { "name": "search", "arguments": {"query": "example"} } ) time.sleep(2) # Fetch test (auto detect file) file_id = list_vector_store_files() if file_id: send_mcp_message( messages, "call_tool", { "name": "fetch", "arguments": {"id": file_id} } ) else: print("\n⚠️ No vector store files → skipping fetch test") # Keep listening a bit longer print("\n⌛ Waiting for responses...") time.sleep(10) if __name__ == "__main__": run_test()