Function main

AskAgent/python/testing/check_connectivity.py:129–190

()

async def main():
    # Create and run all checks simultaneously
    tasks = []

    # Parse command line arguments
    parser = argparse.ArgumentParser(description="Check connectivity for all services required by NLWeb")
    parser.add_argument("--all", action='store_true', default=False,
                        help="Run all connectivity checks for every known provider",)
    args = parser.parse_args()

    start_time = time.time()

    if args.all:
        """Run all connectivity checks"""
        print("Running NLWeb connectivity checks for all known providers...")
        print("This may take a while, please be patient...")
        for llm_provider in CONFIG.llm_endpoints:
            tasks.append(check_llm_api(llm_provider))

        for embedding_provider in CONFIG.embedding_providers:
            tasks.append(check_embedding_api(embedding_provider))

        for retrieval_provider in CONFIG.retrieval_endpoints:
            tasks.append(check_retriever(retrieval_provider))

    else:
        """Run connectivity checks for preferred providers only"""
        print("Checking NLWeb configuration and connectivity...")

        # Retrieve preferred provider from config
        model_config = CONFIG.preferred_llm_endpoint
        print(f"Using configuration from preferred LLM provider: {model_config}")
        tasks.append(check_llm_api(model_config))

        embedding_config = CONFIG.preferred_embedding_provider
        print(f"Using configuration from preferred embedding provider: {embedding_config}")
        tasks.append(check_embedding_api(embedding_config))

        # NOTE: I can't use a retrieval client.enabled_endpoints here, because it does validation and will just remove any invalid endpoints.
        # The purpose of this method is to surface these invalid endpoints to the user.
        for retrieval_endpoint in CONFIG.retrieval_endpoints:
            if CONFIG.retrieval_endpoints[retrieval_endpoint].enabled:
                print(f"Using configuration from enabled retrieval endpoint: {retrieval_endpoint}")
                tasks.append(check_retriever(retrieval_endpoint))

    # Run all tasks concurrently
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Count successful connections
    successful = sum(1 for r in results if r is True)
    total = len(tasks)

    print("\n====== SUMMARY ======")
    print(f"✅ {successful}/{total} connections successful")

    if successful < total:
        print("❌ Some connections failed. Please check error messages above.")
    else: