MCP Image Recognition Server
by mario-andreschak
Verified
import logging
import os
from typing import Optional
from openai import (APIConnectionError, APIError, APITimeoutError, AsyncOpenAI,
RateLimitError)
logger = logging.getLogger(__name__)
class OpenAIVision:
def __init__(self, api_key: Optional[str] = None):
"""Initialize OpenAI Vision client.
Args:
api_key: Optional API key. If not provided, will try to get from environment.
"""
self.api_key = api_key or os.getenv("OPENAI_API_KEY")
if not self.api_key:
raise ValueError("OpenAI API key not provided and not found in environment")
self.base_url = os.getenv("OPENAI_BASE_URL")
timeout_value = os.getenv("OPENAI_TIMEOUT", 60)
self.timeout = float(timeout_value)
self.client = AsyncOpenAI(
api_key=self.api_key, base_url=self.base_url, timeout=self.timeout
)
async def describe_image(
self,
image: str,
prompt: str = "Please describe this image in detail.",
mime_type="image/png",
) -> str:
"""Describe an image using OpenAI's GPT-4 Vision.
Args:
image: String containing base64 encoded image.
prompt: String containing the prompt.
Returns:
str: Description of the image
Raises:
Exception: If API call fails
"""
try:
# Get model from environment, default to gpt-4o-mini
model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
# Create message content
response = await self.client.chat.completions.create(
model=model,
messages=[
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{image}"
},
},
{"type": "text", "text": prompt},
],
}
],
max_tokens=1024,
)
# Extract and return description
return response.choices[0].message.content or "No description available."
except APITimeoutError as e:
logger.error(f"OpenAI API timeout: {str(e)}")
raise Exception(f"Request timed out: {str(e)}")
except APIConnectionError as e:
logger.error(f"OpenAI API connection error: {str(e)}")
raise Exception(f"Connection error: {str(e)}")
except RateLimitError as e:
logger.error(f"OpenAI API rate limit exceeded: {str(e)}")
raise Exception(f"Rate limit exceeded: {str(e)}")
except APIError as e:
logger.error(f"OpenAI API error: {str(e)}")
raise Exception(f"API error: {str(e)}")
except Exception as e:
logger.error(f"Unexpected error in OpenAI Vision: {str(e)}", exc_info=True)
raise Exception(f"Unexpected error: {str(e)}")