lyzr-cortex-sdk 0.1.4

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)

Meeting Intelligence Example

A common pattern: push transcripts after each meeting, then query Cortex to answer user questions — no separate search database needed.

from cortex_sdk import CortexClient

cortex = CortexClient()

# 1. After a meeting ends — push the transcript
await cortex.push(
    type="meeting_transcript",
    id=f"meeting_{meeting_id}_{user_email}",   # unique per user so each person gets their own copy
    name="Q2 Planning — Product Team",
    content="Alice: We need to ship the dashboard by end of April. Bob: I'll own backend...",
    scope="personal",
    metadata={
        "user_email": user_email,
        "meeting_id": meeting_id,
        "participants": ["[e-mail]", "[e-mail]"],
    }
)

# 2. When user asks a question inside your tool — query Cortex
result = await cortex.query(
    question="What did I commit to in recent planning meetings?",
    filters={"type": ["meeting_transcript"]},
    user_email=user_email,     # scopes results to this user's personal docs only
    time_range_days=30,
)

print(result.answer)
# → "In the Q2 Planning meeting you committed to owning the frontend for
#    the new dashboard, targeting end of April."

print([s["name"] for s in result.sources])
# → ["Q2 Planning — Product Team"]

Tip: Pass user_email in both the push() metadata and query() call. Cortex uses it to enforce personal-scope isolation — each user can only retrieve their own transcripts even though they all share the same tool API key.

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 Endpoints

The SDK routes all requests to the Cortex server automatically. CORTEX_API_URL should be the base server URL (e.g., https://internal-auto-ogiserver.lyzr.app). The SDK prepends /api to all paths.

All requests include authentication headers:

X-Cortex-Tool-ID: {CORTEX_TOOL_ID}
X-API-Key: {CORTEX_API_KEY}

SDK Method	HTTP Endpoint	Description
push()	POST /api/tools/documents	Push document to knowledge graph
query()	POST /api/tools/query	Query the knowledge graph
share_document()	POST /api/tools/share-document-for-user	Share document with a user
get_user_context()	GET /api/tools/context/{user_email}	Get user info and teams
get_org_users()	GET /api/tools/org-users-for-tool	List organization users

Push Request Body

POST /api/tools/documents

{
  "external_id": "[e-mail]",
  "document_type": "meeting_transcript",
  "content": "Alice: Hello everyone...",
  "access_scope": "personal",
  "name": "Product Sync - Jan 15",
  "metadata": {
    "user_email": "[e-mail]",
    "meeting_id": "cal_abc123",
    "participants": ["[e-mail]", "[e-mail]"]
  }
}

Share Request Body

POST /api/tools/share-document-for-user

{
  "document_external_id": "[e-mail]",
  "shared_by_email": "[e-mail]",
  "shared_with_email": "[e-mail]",
  "document_name": "Product Sync - Jan 15",
  "content": "Alice: Hello everyone...",
  "metadata": {"document_type": "meeting_transcript"}
}

Query Request Body

POST /api/tools/query

{
  "question": "What decisions were made about Q1?",
  "filters": {"type": ["meeting_transcript"]},
  "time_range_days": 30,
  "max_results": 10,
  "user_email": "[e-mail]"
}

Where Documents Appear in Cortex UI

Access Scope	Cortex Route	Description
personal	/personal/transcripts	Owner's "My Transcripts"
team	Team knowledge base	Visible to team members
global	Organization knowledge base	Visible to everyone
Shared via share_document()	/personal/shared	Recipient's "Shared with Me"

---

API Reference

CortexClient

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

Push a document to the Knowledge Graph. Calls POST /api/tools/documents.

• type: Document type (e.g., "meeting_transcript")
• id: Unique identifier (external_id in Cortex)
• 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 (include user_email for personal scope)

Returns DocumentPushResponse with id and status.

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

Query the Knowledge Graph. Calls POST /api/tools/query.

• 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. Calls GET /api/tools/context/{user_email}.

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. Calls POST /api/tools/share-document-for-user.

• 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. Calls GET /api/tools/org-users-for-tool.

• 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