wehub-resource-sync
c889a57b6b
Test Suites / Build CI Environment (push) Has been cancelled
Test Suites / Basic Tests (push) Has been cancelled
Test Suites / End-to-End Tests (push) Has been cancelled
Test Suites / CLI Tests (push) Has been cancelled
Test Suites / Slow End-to-End Tests (push) Has been cancelled
Test Suites / Graph Database Tests (push) Has been cancelled
Test Suites / Vector DB Tests (push) Has been cancelled
Test Suites / Temporal Graph Test (push) Has been cancelled
Test Suites / Search Test on Different DBs (push) Has been cancelled
Test Suites / Example Tests (push) Has been cancelled
Test Suites / Notebook Tests (push) Has been cancelled
Test Suites / OS and Python Tests Ubuntu (push) Has been cancelled
Test Suites / OS and Python Tests Extended (push) Has been cancelled
Test Suites / LLM Test Suite (push) Has been cancelled
Test Suites / S3 File Storage Test (push) Has been cancelled
Test Suites / Run Integration Tests (push) Has been cancelled
Test Suites / MCP Tests (push) Has been cancelled
Test Suites / Docker Compose Test (push) Has been cancelled
Test Suites / Docker CI test (push) Has been cancelled
Test Suites / Relational DB Migration Tests (push) Has been cancelled
Test Suites / Distributed Cognee Test (push) Has been cancelled
Test Suites / DB Examples Tests (push) Has been cancelled
Test Suites / Test Completion Status (push) Has been cancelled
Test Suites / Claude Code Review (push) Has been cancelled
Test Suites / basic checks (push) Has been cancelled
build | Build and Push Cognee MCP Docker Image to dockerhub / docker-build-and-push (push) Has been cancelled
Scorecard supply-chain security / Scorecard analysis (push) Has been cancelled
build | Build and Push Docker Image to dockerhub / docker-build-and-push (push) Has been cancelled
Weighted Edges Tests / Test Weighted Edges Core Functionality (3.11) (push) Has been cancelled
Weighted Edges Tests / Test Weighted Edges Core Functionality (3.12) (push) Has been cancelled
Weighted Edges Tests / Test Weighted Edges with Different Graph Databases (kuzu, kuzu) (push) Has been cancelled
Weighted Edges Tests / Test Weighted Edges with Different Graph Databases (neo4j, neo4j) (push) Has been cancelled
Weighted Edges Tests / Test Weighted Edges Examples (push) Has been cancelled
Weighted Edges Tests / Code Quality for Weighted Edges (push) Has been cancelled
74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
import argparse
|
|
import asyncio
|
|
|
|
from cognee.cli.reference import SupportsCliCommand
|
|
from cognee.cli import DEFAULT_DOCS_URL
|
|
import cognee.cli.echo as fmt
|
|
|
|
|
|
class ServeCommand(SupportsCliCommand):
|
|
command_string = "serve"
|
|
help_string = "Connect to a Cognee instance (cloud or local)"
|
|
docs_url = DEFAULT_DOCS_URL
|
|
description = """
|
|
Connect the local Cognee SDK to a Cognee instance.
|
|
|
|
Cloud mode (default): authenticates via browser-based device code flow,
|
|
discovers your tenant, and connects automatically.
|
|
|
|
Local mode: connect directly to a running Cognee backend.
|
|
|
|
Examples:
|
|
cognee serve # Cloud (Auth0 device flow)
|
|
cognee serve --url http://localhost:8000 # Local instance
|
|
cognee serve --url https://my.cognee.ai --api-key ck_...
|
|
cognee serve --logout # Disconnect + clear credentials
|
|
"""
|
|
|
|
def configure_parser(self, parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument(
|
|
"--url",
|
|
help="Direct URL of a Cognee instance (skips Auth0 + tenant discovery)",
|
|
)
|
|
parser.add_argument(
|
|
"--api-key",
|
|
help="API key for the instance (optional for local, required for cloud instances)",
|
|
)
|
|
parser.add_argument(
|
|
"--management-url",
|
|
help="Override the Management API URL (cloud mode only)",
|
|
)
|
|
parser.add_argument(
|
|
"--logout",
|
|
action="store_true",
|
|
help="Disconnect and clear saved credentials",
|
|
)
|
|
|
|
def execute(self, args: argparse.Namespace) -> None:
|
|
if args.logout:
|
|
asyncio.run(_logout())
|
|
else:
|
|
asyncio.run(
|
|
_serve(url=args.url, api_key=args.api_key, management_url=args.management_url)
|
|
)
|
|
|
|
|
|
async def _serve(url=None, api_key=None, management_url=None):
|
|
import cognee
|
|
|
|
mode = "local" if url else "cloud"
|
|
fmt.note(f"Connecting to Cognee ({mode})...")
|
|
try:
|
|
client = await cognee.serve(url=url, api_key=api_key or "", management_url=management_url)
|
|
fmt.success(f"Connected to {client.service_url}")
|
|
except KeyboardInterrupt:
|
|
fmt.warning("Authentication cancelled.")
|
|
except Exception as e:
|
|
fmt.error(f"Failed to connect: {e}")
|
|
|
|
|
|
async def _logout():
|
|
import cognee
|
|
|
|
await cognee.disconnect(clear_saved=True)
|