AI API Manager Extension

Proposing API integration

I'm considering how to enrich AIAPIManager by introducing a user-friendly menu. This menu would facilitate adding desired APIs, complete with documentation and keys, ensuring ease of use.

Projecting functionality

Integrating a Spotify API setting, users would input commands like reading playlists. The AI would then generate and test the corresponding code, ensuring functionality and accuracy.

Enhancing functionality

By integrating a advanced method to generate code from documentation, adding a dynamic API menu for creating and updating APIs, and refining the code execution process, the system's functionality will be further enhanced.

Thought about API settings and automation pendant 20 secondes

Below is an extended version of your code that implements a Settings menu. This settings menu allows the user to:

1. Add new APIs manually (by providing a base URL, authentication, and so on).
2. Provide documentation for new APIs or commands.
3. Delegate to the AI the task of parsing that documentation to generate endpoints (just like your smart_api_analysis function).
4. Validate or test the newly created endpoints.

To illustrate, I added an example flow for creating a Spotify API entry. The user can provide minimal documentation or a short text describing what they want to do (e.g. "Read my playlists from Spotify"). The AI will try to parse it and generate endpoints automatically. Then the user can test them.

Note:

• The below code keeps your existing code structure but adds new menu items and methods to handle advanced “settings” scenarios.
• For a real-world Spotify integration, you would need OAuth flow, which can get more complex, but we show a simplified approach here to demonstrate the idea of adding a new API with a custom command.
• If you really want to call Spotify, you might need to handle OAuth tokens, refresh tokens, etc. This code is a skeleton for demonstration.

import json
import aiohttp
import asyncio
from groq import AsyncGroq
from typing import Dict, Optional, List
import logging
from datetime import datetime
import re
from urllib.parse import urlparse, parse_qs

class AIAPIManager:
    def __init__(self, settings_file='api_settings.json', groq_api_key='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'):
        self.settings_file = settings_file
        self.apis = self.load_settings()
        self.groq_client = AsyncGroq(api_key=groq_api_key)
        self.logger = self._setup_logger()

    def _setup_logger(self):
        logger = logging.getLogger('AIAPIManager')
        logger.setLevel(logging.INFO)
        if not logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
            handler.setFormatter(formatter)
            logger.addHandler(handler)
        return logger

    def load_settings(self):
        try:
            with open(self.settings_file, 'r') as file:
                return json.load(file)
        except FileNotFoundError:
            return {}

    def save_settings(self):
        with open(self.settings_file, 'w') as file:
            json.dump(self.apis, file, indent=4)

    async def smart_api_analysis(self, user_input: str) -> Dict:
        """Analyse intelligente de l'API à partir de l'entrée utilisateur"""
        message = f"""You are an API configuration expert. Analyze this user input and create a detailed API configuration.
If it's a URL, analyze its structure. If it's an API description, extract key information.

User Input: {user_input}

Consider these aspects:
1. Is this a URL or API description?
2. What's the authentication method?
3. What's the API's purpose?
4. What are the common endpoints?
5. What parameters are typically required?

Respond in this exact JSON format:
{{
    "api_type": "rest|graphql|soap",
    "base_url": "string",
    "auth_info": {{
        "type": "none|api_key|oauth|bearer|basic",
        "location": "none|header|query|basic",
        "key_name": "string",
        "required": true|false
    }},
    "common_headers": {{"key": "value"}},
    "detected_endpoints": [
        {{
            "name": "string",
            "path": "string",
            "method": "GET|POST|PUT|DELETE",
            "required_params": ["param1", "param2"],
            "optional_params": ["param3"],
            "description": "string"
        }}
    ],
    "suggested_name": "string",
    "rate_limit_info": "string",
    "confidence_score": 0.0
}}"""

        try:
            chat_completion = await self.groq_client.chat.completions.create(
                messages=[
                    {"role": "user", "content": message}
                ],
                model="llama-3.1-8b-instant",
                temperature=0.1,
                stream=False
            )

            response_content = chat_completion.choices[0].message.content
            # Handle potential JSONDecodeError
            try:
                return json.loads(response_content)
            except json.JSONDecodeError:
                self.logger.error("Erreur Groq: Invalid JSON response. Falling back to basic analysis.")
                return await self.fallback_analysis(user_input)

        except Exception as e:
            self.logger.error(f"Erreur Groq: {str(e)}")
            return await self.fallback_analysis(user_input)

    async def fallback_analysis(self, user_input: str) -> Dict:
        """Analyse de repli si Groq n'est pas disponible ou si la réponse JSON est invalide"""
        try:
            parsed = urlparse(user_input)
            # If scheme and netloc are present, it is likely a URL
            if parsed.scheme and parsed.netloc:
                return self.analyze_url(parsed)
        except Exception as e:
            print(e)
        # If not a URL or other error, create a very basic config with low confidence
        return self.create_basic_config(user_input)

    def create_basic_config(self, user_input: str) -> Dict:
        """Crée une configuration de base avec des suggestions d'endpoints si possible"""
        config = {
            "api_type": "rest",
            "base_url": "",
            "auth_info": {
                "type": "none",
                "location": "none",
                "key_name": "",
                "required": False
            },
            "common_headers": {},
            "detected_endpoints": [],
            "suggested_name": "unknown_api",
            "rate_limit_info": "unknown",
            "confidence_score": 0.2
        }

        # Basic endpoint suggestions based on user input
        user_input_lower = user_input.lower()
        if "cat" in user_input_lower or "chat" in user_input_lower:
            config["detected_endpoints"].append({
                "name": "get_cats",
                "path": "/cats",
                "method": "GET",
                "required_params": [],
                "optional_params": [],
                "description": "Get a list of cats"
            })
            config["detected_endpoints"].append({
                "name": "get_cat",
                "path": "/cats/{id}",
                "method": "GET",
                "required_params": ["id"],
                "optional_params": [],
                "description": "Get a specific cat by ID"
            })
            config["suggested_name"] = "cat_api"

        elif "nasa" in user_input_lower:
            config["detected_endpoints"].append({
                "name": "get_apod",
                "path": "/planetary/apod",
                "method": "GET",
                "required_params": [],
                "optional_params": ["date", "api_key"],
                "description": "Get Astronomy Picture of the Day"
            })
            config["suggested_name"] = "nasa_api"
            config["base_url"] = "https://api.nasa.gov"

        return config

    def analyze_url(self, parsed_url) -> Dict:
        """Analyse une URL pour créer une configuration"""
        params = parse_qs(parsed_url.query)
        base_url = f"{parsed_url.scheme}://{parsed_url.netloc}"

        auth_info = {
            "type": "none",
            "location": "none",
            "key_name": "",
            "required": False
        }

        for param in params:
            if any(auth_term in param.lower() for auth_term in ['api_key', 'apikey', 'key', 'token']):
                auth_info = {
                    "type": "api_key",
                    "location": "query",
                    "key_name": param,
                    "required": True
                }
                break

        return {
            "api_type": "rest",
            "base_url": base_url,
            "auth_info": auth_info,
            "common_headers": {},
            "detected_endpoints": [{
                "name": "default",
                "path": parsed_url.path,
                "method": "GET",
                "required_params": list(params.keys()),
                "optional_params": [],
                "description": "Default endpoint detected from URL"
            }],
            "suggested_name": parsed_url.netloc.split('.')[0],
            "rate_limit_info": "unknown",
            "confidence_score": 0.7
        }

    async def smart_add_api(self):
        """Interface utilisateur intelligente pour l'ajout d'API"""
        print("\n🤖 Assistant IA d'intégration d'API\n")
        print("Entrez l'URL de l'API ou décrivez-la en langage naturel:")
        user_input = input().strip()

        print("\n⏳ Analyse en cours...")
        config = await self.smart_api_analysis(user_input)

        if config['confidence_score'] < 0.5:
            print("\n⚠️ L'analyse automatique n'est pas très sûre. Aidons l'IA à mieux comprendre.")
            # Guide the user for more information
            if not config['base_url']:
                config['base_url'] = input("Avez-vous une URL de base pour l'API ? (Entrée pour ignorer) : ").strip()
            if config['auth_info']['type'] == 'none':
                auth_input = input("L'API nécessite-t-elle une authentification (o/n) ? ").strip().lower()
                if auth_input == 'o':
                    config['auth_info']['type'] = input("Quel type d'authentification (api_key, oauth, bearer, basic) ? ").strip()
                    config['auth_info']['location'] = input("Où se trouve la clé d'authentification (header, query, basic) ? ").strip()
                    config['auth_info']['key_name'] = input("Quel est le nom du paramètre ou de l'en-tête pour l'authentification ? ").strip()
                    config['auth_info']['required'] = True

        print(f"\n📝 Configuration détectée (confiance: {config['confidence_score']*100}%):")
        print(f"Type d'API: {config['api_type']}")
        print(f"Base URL: {config['base_url']}")
        print(f"Authentification: {config['auth_info']['type']}")

        suggested_name = config['suggested_name']
        print(f"\nNom suggéré pour l'API: {suggested_name}")
        api_name = input("Appuyez sur Entrée pour accepter ou entrez un nouveau nom: ").strip()
        api_name = api_name or suggested_name

        if config['auth_info']['required']:
            print(f"\nAuthentification requise ({config['auth_info']['type']}):")
            auth_value = input(f"Entrez votre {config['auth_info']['key_name']}: ").strip()
            config['auth_info']['value'] = auth_value

        # Store new API config
        self.apis[api_name] = {
            "base_url": config['base_url'],
            "auth_info": config['auth_info'],
            "endpoints": {endpoint['name']: endpoint for endpoint in config['detected_endpoints']},
            "headers": config['common_headers'],
            "last_updated": datetime.now().isoformat()
        }

        self.save_settings()
        print(f"\n✅ API '{api_name}' configurée avec succès!")

        # Proposer un test
        if input("\nVoulez-vous tester un endpoint maintenant? (o/n): ").lower() == 'o':
            if not config['detected_endpoints']:
                 print("\nℹ️ Aucun endpoint détecté automatiquement. Vous pouvez en ajouter un manuellement pour le test.")
            await self.test_endpoint(api_name)

    async def test_endpoint(self, api_name: str):
        """Test rapide d'un endpoint"""
        api_config = self.apis[api_name]
        endpoints = api_config.get('endpoints', {})

        if not endpoints:
            print("\nℹ️ Aucun endpoint disponible pour le test.")
            add_manually = input("Voulez-vous ajouter un endpoint manuellement pour le tester (o/n) ? ").lower()
            if add_manually == 'o':
                endpoint_name = input("Nom du endpoint : ")
                endpoint_path = input("Chemin du endpoint (ex: /users) : ")
                endpoint_method = input("Méthode HTTP (GET, POST, PUT, DELETE) : ").upper()
                endpoints[endpoint_name] = {
                    "name": endpoint_name,
                    "path": endpoint_path,
                    "method": endpoint_method,
                    "required_params": [],
                    "optional_params": [],
                    "description": "Endpoint ajouté manuellement"
                }
                # Save the manually added endpoint
                self.apis[api_name]['endpoints'] = endpoints
                self.save_settings()
            else:
                return

        print("\nEndpoints disponibles:")
        for name, endpoint in endpoints.items():
            print(f"- {name}: {endpoint['method']} {endpoint['path']}")

        endpoint_name = input("\nChoisissez un endpoint à tester: ").strip()
        if endpoint_name not in endpoints:
            print("❌ Endpoint non trouvé")
            return

        endpoint = endpoints[endpoint_name]
        params = {}

        for param in endpoint['required_params']:
            if param != api_config['auth_info'].get('key_name'):
                params[param] = input(f"Entrez {param}: ").strip()

        try:
            result = await self.call_endpoint(api_name, endpoint_name, params)
            print("\n✅ Résultat du test:")
            print(json.dumps(result, indent=2))
        except Exception as e:
            print(f"\n❌ Erreur lors du test: {str(e)}")

    async def call_endpoint(self, api_name: str, endpoint_name: str, params: Dict) -> Optional[Dict]:
        """Exécute un appel API"""
        api_config = self.apis[api_name]
        endpoint = api_config['endpoints'][endpoint_name]

        url = api_config['base_url'] + endpoint['path']

        headers = api_config['headers'].copy()
        if api_config['auth_info']['type'] != 'none':
            # example: handle Bearer tokens, basic, etc. if needed
            if api_config['auth_info']['type'] in ['bearer', 'oauth']:
                headers['Authorization'] = f"Bearer {api_config['auth_info']['value']}"
            elif api_config['auth_info']['type'] == 'basic':
                # For BASIC, user might have "username:password"
                # This is a simplistic approach - in real usage we'd base64-encode
                headers['Authorization'] = f"Basic {api_config['auth_info']['value']}"
            elif api_config['auth_info']['type'] == 'api_key':
                if api_config['auth_info']['location'] == 'header':
                    headers[api_config['auth_info']['key_name']] = api_config['auth_info']['value']
                elif api_config['auth_info']['location'] == 'query':
                    params[api_config['auth_info']['key_name']] = api_config['auth_info']['value']

        async with aiohttp.ClientSession() as session:
            method = getattr(session, endpoint['method'].lower())
            async with method(url, headers=headers, params=params) as response:
                response.raise_for_status()
                # If response is not JSON, this might fail:
                return await response.json()

    # ------------------ NEW SETTINGS METHODS ------------------ #

    async def manage_settings(self):
        """
        Manage user settings:
        1. Add a new API (manually or with AI help)
        2. Update or remove an API
        3. Add new commands to an existing API from user-provided docs
        """
        while True:
            print("\n⚙️  Paramètres")
            print("1. Ajouter une nouvelle API (assisté par IA)")
            print("2. Mettre à jour / Supprimer une API existante")
            print("3. Ajouter de nouvelles commandes/endpoints via documentation")
            print("4. Retour au menu principal")

            choice = input("\nChoix: ").strip()

            if choice == '1':
                # Re-use our existing smart_add_api
                await self.smart_add_api()
            elif choice == '2':
                self.update_or_remove_api()
            elif choice == '3':
                if not self.apis:
                    print("\n❌ Aucune API configurée pour le moment.")
                    continue
                await self.add_commands_via_docs()
            elif choice == '4':
                break
            else:
                print("\n❌ Choix invalide")

    def update_or_remove_api(self):
        """Update or remove an existing API from the config"""
        if not self.apis:
            print("\n❌ Aucune API configurée.")
            return

        print("\nAPIs disponibles:")
        for name in self.apis.keys():
            print(f"- {name}")

        api_name = input("\nChoisissez une API à modifier/supprimer: ").strip()
        if api_name not in self.apis:
            print("❌ API non trouvée.")
            return

        print(f"\nQue voulez-vous faire avec l'API '{api_name}' ?")
        print("1. Mettre à jour la configuration (base URL, auth, etc.)")
        print("2. Supprimer l'API")
        choice = input("\nChoix: ").strip()

        if choice == '1':
            self.update_api_config(api_name)
        elif choice == '2':
            del self.apis[api_name]
            self.save_settings()
            print(f"\n✅ L'API '{api_name}' a été supprimée.")
        else:
            print("\n❌ Choix invalide")

    def update_api_config(self, api_name: str):
        """Update basic info like base_url, auth_info, etc."""
        api_config = self.apis[api_name]
        print("\nLaissez vide pour conserver les valeurs actuelles.\n")

        new_base_url = input(f"Base URL actuelle: {api_config['base_url']}\nNouvelle base URL: ").strip()
        if new_base_url:
            api_config['base_url'] = new_base_url

        print(f"\nAuthentification actuelle: {api_config['auth_info']['type']}")
        update_auth = input("Voulez-vous changer l'authentification (o/n)? ").strip().lower()
        if update_auth == 'o':
            auth_type = input("Nouveau type (api_key, oauth, bearer, basic, none): ").strip()
            auth_location = input("Location (header, query, basic, none): ").strip()
            key_name = input("Nom du paramètre ou en-tête: ").strip()
            required_val = input("Requis ? (o/n): ").strip().lower()

            if required_val == 'o':
                required = True
                auth_value = input(f"Entrez votre {key_name}: ").strip()
            else:
                required = False
                auth_value = ""

            api_config['auth_info'] = {
                "type": auth_type,
                "location": auth_location,
                "key_name": key_name,
                "required": required,
                "value": auth_value
            }

        self.apis[api_name] = api_config
        self.apis[api_name]['last_updated'] = datetime.now().isoformat()
        self.save_settings()
        print(f"\n✅ Configuration de l'API '{api_name}' mise à jour.")

    async def add_commands_via_docs(self):
        """
        Allows user to select an existing API and provide some 'documentation' or 'description'
        of new commands they want to add. The AI tries to parse it and generate new endpoints.
        """
        print("\nAPIs disponibles:")
        for name in self.apis.keys():
            print(f"- {name}")

        api_name = input("\nChoisissez une API pour ajouter de nouvelles commandes: ").strip()
        if api_name not in self.apis:
            print("❌ API non trouvée.")
            return

        print("\nEntrez la documentation ou la description de la/les commande(s) que vous voulez ajouter.")
        print("Par exemple, pour Spotify: 'Je veux récupérer la liste de mes playlists, récupérer les titres d'une playlist, etc.'")
        docs_input = input("\nVotre description: ")

        if not docs_input:
            print("\n❌ Description vide.")
            return

        print("\n⏳ Analyse de votre description en cours...")
        new_config = await self.smart_api_analysis(docs_input)

        # If the AI found new endpoints, let's merge them:
        if new_config.get('detected_endpoints'):
            existing_endpoints = self.apis[api_name].get('endpoints', {})
            for endpoint in new_config['detected_endpoints']:
                ep_name = endpoint['name']
                # If name already exists, append a suffix
                while ep_name in existing_endpoints:
                    ep_name = ep_name + "_new"
                existing_endpoints[ep_name] = endpoint

            self.apis[api_name]['endpoints'] = existing_endpoints
            self.apis[api_name]['last_updated'] = datetime.now().isoformat()

            # Possibly update base_url or other info if not present
            if not self.apis[api_name]['base_url'] and new_config.get('base_url'):
                self.apis[api_name]['base_url'] = new_config['base_url']

            # We won't override existing auth_info, but we can compare
            if new_config.get('auth_info', {}).get('type') != 'none' and \
               new_config['auth_info']['type'] != self.apis[api_name]['auth_info']['type']:
                print("\n⚠️ Nouvelle info d'auth détectée. Souhaitez-vous la remplacer ? (o/n)")
                if input().strip().lower() == 'o':
                    self.apis[api_name]['auth_info'] = new_config['auth_info']

            self.save_settings()
            print("\n✅ Nouvelles commandes ajoutées avec succès !")
        else:
            print("\n❌ Aucune commande détectée dans la documentation. Vous pouvez réessayer avec plus de détails.")

    # ------------------ MAIN MENU ------------------ #

def main():
    manager = AIAPIManager(groq_api_key="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")  # Replace with your own Groq API key

    async def run_menu():
        while True:
            print("\n🔧 Menu Principal")
            print("1. 🤖 Ajouter une API (assisté par IA)")
            print("2. 📋 Liste des APIs")
            print("3. 🔄 Tester un endpoint")
            print("4. ⚙️  Paramètres (ajout/suppression/modification d'APIs, nouvelles commandes)")
            print("5. 🚪 Quitter")

            choice = input("\nChoix: ").strip()

            if choice == '1':
                # Directly add an API
                await manager.smart_add_api()
            elif choice == '2':
                # List all APIs
                if not manager.apis:
                    print("\n❌ Aucune API configurée.")
                else:
                    for name, config in manager.apis.items():
                        print(f"\n📌 {name}")
                        print(f"Base URL: {config['base_url']}")
                        print("Endpoints:")
                        for endpoint in config['endpoints'].values():
                            print(f"  - {endpoint['method']} {endpoint['path']}")
            elif choice == '3':
                if not manager.apis:
                    print("\n❌ Aucune API configurée")
                    continue
                print("\nAPIs disponibles:")
                for name in manager.apis:
                    print(f"- {name}")
                api_name = input("\nChoisissez une API: ").strip()
                if api_name in manager.apis:
                    await manager.test_endpoint(api_name)
                else:
                    print("\n❌ API non trouvée.")
            elif choice == '4':
                # Manage settings
                await manager.manage_settings()
            elif choice == '5':
                print("\n👋 Au revoir!")
                break
            else:
                print("\n❌ Choix invalide")

    asyncio.run(run_menu())

if __name__ == "__main__":
    main()

What’s New?

1. New “Settings” Menu (manage_settings method):

   • 1. Ajouter une nouvelle API: re-uses your existing AI-driven flow for adding an API.
   • 2. Mettre à jour / Supprimer une API: update or remove an existing API from the configuration.
   • 3. Ajouter de nouvelles commandes/endpoints via documentation: the user can provide new documentation/descriptions, and the AI tries to parse and generate endpoints.

2. Adding or Updating an API:

   • In the “update” flow, we allow the user to change the base_url and authentication details.
   • In the “remove” flow, we simply delete the selected API from the config.

3. Generate Commands from Documentation:

   • The user can input custom docs (like “I want to fetch Spotify playlists”), and the same AI logic (smart_api_analysis) attempts to generate endpoints.
   • We merge these newly discovered endpoints with existing ones.

4. Bearer, OAuth, Basic Auth handling**:

   • In call_endpoint, we expanded the logic to handle different authentication styles in a basic way (though real OAuth flows may require a separate token request).

With this, you have a more “instinctive” tool to add or remove APIs, add custom commands, test them, and store everything in a local JSON file. You can keep building on this to incorporate real OAuth flows for something like Spotify or other complex APIs.