lyzr-cortex-sdk 0.1.0

SDK for building tools that integrate with Lyzr Cortex Platform

Cortex SDK for Python

Build tools that integrate with the Cortex Platform for bidirectional knowledge flow.

Installation

pip install cortex-sdk

# With FastAPI support
pip install cortex-sdk[fastapi]

Quick Start

from cortex_sdk import CortexClient

# Auto-configured from environment variables
client = CortexClient()

# Push a document to Knowledge Graph
await client.push(
    type="meeting_transcript",
    id="transcript_123",
    content="Alice: Let's discuss the Q1 roadmap...",
    name="Product Sync - Jan 15",
    scope="team",
    teams=["team_product"],
    metadata={
        "participants": ["[e-mail]", "[e-mail]"],
        "duration_minutes": 45
    }
)

# Query the Knowledge Graph
result = await client.query(
    question="What decisions were made about Q1?",
    filters={"type": ["meeting_transcript"]},
    time_range_days=30
)
print(result.answer)
print(result.sources)

Environment Variables

Variable	Description	Required
CORTEX_ENABLED	Enable/disable integration	No (default: true)
CORTEX_API_URL	Cortex gateway URL	Yes
CORTEX_API_KEY	Tool's API key	Yes
CORTEX_TOOL_ID	Tool identifier	Yes
CORTEX_JWT_PUBLIC_KEY	RSA public key for JWT validation	For embedded mode

FastAPI Integration

from fastapi import FastAPI, Depends
from cortex_sdk import CortexClient
from cortex_sdk.auth import get_current_cortex_user, require_cortex_user
from cortex_sdk.models import CortexUser

app = FastAPI()
cortex = CortexClient()

@app.get("/data")
async def get_data(user: CortexUser | None = Depends(get_current_cortex_user)):
    """Works in both standalone and embedded mode."""
    if user:
        # Embedded mode - use Cortex context
        result = await cortex.query(
            f"Get data for {user.email}",
            user_email=user.email
        )
        return {"data": result.answer, "user": user.email}

    # Standalone mode
    return {"data": "default data"}

@app.get("/protected")
async def protected_endpoint(user: CortexUser = Depends(require_cortex_user)):
    """Only works in embedded mode - requires Cortex auth."""
    return {"email": user.email, "teams": user.team_names}

Graceful Degradation

The SDK operates in no-op mode when Cortex is not available:

client = CortexClient()

if client.enabled:
    # Cortex is available - full functionality
    await client.push(...)
else:
    # Cortex disabled - operations are no-ops
    result = await client.push(...)  # Returns None
    result = await client.query(...)  # Returns empty result

Access Scopes

Scope	Who Can See	Use For
global	Everyone in organization	Announcements, shared resources
team	Specified team members	Projects, deals, team docs
personal	Only the creator	Notes, drafts, personal items

API Reference

CortexClient

push(type, id, content, scope="global", teams=None, metadata=None)

Push a document to the Knowledge Graph.

• type: Document type (e.g., "meeting_transcript")
• id: Unique identifier
• content: Text content for RAG indexing
• scope: "global", "team", or "personal"
• teams: Team IDs if scope is "team"
• metadata: Additional metadata dict

query(question, filters=None, time_range_days=None, max_results=10)

Query the Knowledge Graph.

• question: Natural language query
• filters: Filter dict (e.g., {"type": ["meeting_transcript"]})
• time_range_days: Limit to recent documents
• max_results: Maximum results

Returns CortexQueryResult with answer and sources.

get_user_context(user_email)

Get user context for access control.

Returns CortexUser with teams, permissions, etc.

License

MIT