src/chat_ai.py

#!/opt/homebrew/bin/python3.12

import os
import sys
from qwen_api import Qwen
from qwen_api.client import QwenAPIError
from qwen_api.core.exceptions import AuthError, RateLimitError
from qwen_api.core.types.chat import ChatMessage, ChatResponse, MessageRole

LLM_LOGIN_USER = os.getenv('LLM_LOGIN_USER','')
LLM_LOGIN_PASSWORD = os.getenv('LLM_LOGIN_PASSWORD','')
LLM_MODEL = os.getenv('LLM_MODEL','qwen-max-latest')

def qwen_chat(prompt: str) -> tuple[bool, str]:
    
    response_status = False

    try:
        # Initialize client
        # Full configuration
        client = Qwen(
            
            email=LLM_LOGIN_USER,
            password=LLM_LOGIN_PASSWORD,
            # api_key = token,                    # API authentication token
            # api_key = QWEN_AUTH_TOKEN,        # API authentication token
            # cookie = cookie,                    # Authentication cookie
            base_url = "https://chat.qwen.ai",  # API base URL
            timeout = 600,                      # Request timeout in seconds
            log_level = "ERROR",                # Logging level
            save_logs = False                   # Whether to save logs to file
        )

        # Create a simple chat message
        messages = [ChatMessage(
            role = MessageRole.USER,
            content = prompt,
            web_search = False,
            thinking = False
        )]

        # Get response
        response: ChatResponse = client.chat.create(
            messages=messages,
            model=LLM_MODEL, # type: ignore
            #model="qwen3-coder-30b-a3b-instruct", # qwen3 coder flash
            stream=True
        ) # type: ignore
        
        response_status = True

        content = ""
        # Process the stream
        for chunk in response:
            delta = chunk.choices[0].delta # type: ignore
            content += delta.content

        return response_status, content
        
    except AuthError as e:
        print(f"Authentication failed: {e}")
    except RateLimitError as e:
        print(f"Rate limit exceeded: {e}")
    except QwenAPIError as e:
        print(f"API error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")
    
    return response_status, "Error"

if __name__ == "__main__":
    if (len(sys.argv) <= 1):
        print("No message provided")
        sys.exit(1)
          
    role = "" 

    prompt = role + ' '.join(sys.argv[1:])

    response_ok, result = qwen_chat(prompt)
    
    if response_ok:
        print(result)