lyzr-cortex-sdk 0.1.2

SDK for building tools that integrate with Lyzr Cortex Platform

Lyzr Cortex SDK for Python

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

Installation

pip install lyzr-cortex-sdk

# With FastAPI support
pip install lyzr-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
• name: Display name for the document
• url: URL to view document in source tool
• 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, user_email=None)

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
• user_email: User email for access-scoped queries

Returns CortexQueryResult with answer and sources.

get_user_context(user_email)

Get user context for access control.

Returns CortexUser with teams, permissions, etc.

share_document(document_external_id, shared_by_email, shared_with_email, ...)

Share a document with another user in the organization.

• document_external_id: External ID of the document to share
• shared_by_email: Email of the user sharing
• shared_with_email: Email of the recipient
• document_name: Display name for the shared document
• content: Document content
• external_url: URL to view the document
• metadata: Additional metadata dict

get_org_users(exclude_email=None)

Get list of users in the tool's organization.

• exclude_email: Email to exclude from results (typically the current user)

Returns list of user dicts with id, email, full_name, avatar_url.

Embedded Mode Authentication (Production)

When your tool runs inside a Cortex iframe, the frontend receives the user's OGI JWT via the CORTEX_INIT postMessage. To ensure authentication works in production (where reverse proxies may strip custom headers), your tool should send both custom Cortex headers and the standard Authorization header.

Frontend: Passing Auth Headers

import { useCortex } from '@/lib/cortex';

function useAuthHeaders() {
  const { isEmbedded, user, token } = useCortex();

  const getHeaders = (): Record<string, string> => {
    if (isEmbedded && user) {
      const headers: Record<string, string> = {
        // Custom headers — work locally, may be stripped by production proxies
        'X-Cortex-Embedded': 'true',
        'X-Cortex-User-Email': user.email,
      };
      // Standard header — survives all reverse proxies / CDNs / ALBs
      if (token) {
        headers['Authorization'] = `Bearer ${token}`;
      }
      return headers;
    }
    // Standalone mode: use your own JWT from localStorage
    const jwt = localStorage.getItem('token');
    return jwt ? { Authorization: `Bearer ${jwt}` } : {};
  };

  return { getHeaders };
}

Backend: Accepting Both Auth Modes

from fastapi import Depends, Header, HTTPException
import jwt

async def get_current_user(
    authorization: str = Header(None),
    x_cortex_embedded: str = Header(None),
    x_cortex_user_email: str = Header(None),
):
    # Mode 1: Cortex embedded — custom headers (local / non-proxied)
    if x_cortex_embedded == "true" and x_cortex_user_email:
        return x_cortex_user_email

    # Mode 2: Bearer token — works through all proxies
    if authorization and authorization.startswith("Bearer "):
        token = authorization.split(" ", 1)[1]
        try:
            # Try your tool's own JWT secret first
            payload = jwt.decode(token, YOUR_SECRET_KEY, algorithms=["HS256"])
            return payload["sub"]
        except jwt.InvalidTokenError:
            pass
        try:
            # Fall back to OGI JWT (CORTEX_JWT_PUBLIC_KEY)
            payload = jwt.decode(token, CORTEX_JWT_PUBLIC_KEY, algorithms=["HS256"])
            return payload.get("email") or payload.get("sub")
        except jwt.InvalidTokenError:
            pass

    raise HTTPException(status_code=401, detail="Not authenticated")

Why Both Headers?

Header	Works Locally	Works in Production	Notes
X-Cortex-Embedded	Yes	Maybe	Custom headers may be stripped by CDN/ALB/nginx
X-Cortex-User-Email	Yes	Maybe	Same risk as above
Authorization: Bearer	Yes	Yes	Standard HTTP header, always forwarded

Production flow:

1. User opens your tool inside Cortex iframe
2. Cortex sends CORTEX_INIT with { user, theme, token } via postMessage
3. Your frontend sends API requests with both custom headers AND Authorization: Bearer {token}
4. Your backend tries custom headers first (fast path), falls back to JWT decode (proxy-safe path)

Environment Variables

Variable	Description
CORTEX_JWT_PUBLIC_KEY	OGI's JWT signing key — must match OGI server's JWT_SECRET_KEY

Set this in your tool's .env:

CORTEX_JWT_PUBLIC_KEY=your-ogi-jwt-secret-key

License

MIT