COGNIKIN
GitHub
SDK Reference

Python SDK

Official Python SDK for CogniKin.

Installation

pip install cognikin

Initialisation

from cognikin import CogniKin
import os

brain = CogniKin(
    api_key=os.getenv("COGNIKIN_API_KEY"),

    # Optional configuration
    base_url="https://api.cognikin.me/v1",  # Default
    timeout=5.0,                                   # Request timeout (seconds)
    max_retries=3,                                 # Auto-retry failed requests
    debug=False                                    # Enable debug logging
)

Methods

get_context()

context = await brain.get_context(
    user_id="user_123",
    task="build a feature",
    complexity="medium",      # Optional
    task_type="coding",       # Optional
    metadata={}               # Optional
)

report_outcome()

await brain.report_outcome(
    request_id=context.request_id,
    started=True,
    completed=True,
    time_to_start=45,         # Optional
    flow_state=True,          # Optional
    satisfaction=0.9          # Optional
)

update_profile()

profile = await brain.update_profile(
    user_id="user_123",
    preferred_framing="achievement",
    communication_style="brief_directive"
)

Type Hints

from cognikin import (
    CogniKin,
    KnowledgeContext,
    FramingType,
    CommunicationStyle
)

# Fully typed with Pydantic!
context: KnowledgeContext = await brain.get_context(
    user_id="user_123",
    task="test"
)

Error Handling

from cognikin.exceptions import (
    CogniKinError,
    RateLimitError,
    UnauthorizedError
)

try:
    context = await brain.get_context(user_id=user_id, task=task)
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except UnauthorizedError as e:
    print("Invalid API key")
except CogniKinError as e:
    print(f"CogniKin Error: {e.code} - {e.message}")

Event Hooks

# Register event listeners
@brain.on("request")
async def on_request(params):
    print(f"CogniKin Request: {params}")

@brain.on("response")
async def on_response(context, duration):
    print(f"CogniKin responded in {duration}ms")

@brain.on("error")
async def on_error(error):
    print(f"CogniKin Error: {error}")

Framework Examples

FastAPI

main.py
from fastapi import FastAPI
from cognikin import CogniKin
import os

app = FastAPI()
brain = CogniKin(api_key=os.getenv("COGNIKIN_API_KEY"))

@app.post("/api/agent")
async def handle_query(request: dict):
    context = await brain.get_context(
        user_id=request["userId"],
        task=request["query"]
    )

    return {"context": context.dict()}

Flask

app.py
from flask import Flask, request, jsonify
from cognikin import CogniKin
import asyncio
import os

app = Flask(__name__)
brain = CogniKin(api_key=os.getenv("COGNIKIN_API_KEY"))

@app.route("/api/agent", methods=["POST"])
def handle_query():
    data = request.json

    context = asyncio.run(brain.get_context(
        user_id=data["userId"],
        task=data["query"]
    ))

    return jsonify({"context": context.dict()})

Django

views.py
from django.http import JsonResponse
from cognikin import CogniKin
import os

brain = CogniKin(api_key=os.getenv("COGNIKIN_API_KEY"))

async def agent_view(request):
    import json
    data = json.loads(request.body)

    context = await brain.get_context(
        user_id=data["userId"],
        task=data["query"]
    )

    return JsonResponse({"context": context.dict()})

💡 Async Support: The SDK fully supports both async/await and synchronous usage patterns.

Synchronous Usage

# Synchronous client
from cognikin import CogniKinSync

brain_sync = CogniKinSync(api_key=os.getenv("COGNIKIN_API_KEY"))

# Use without async/await
context = brain_sync.get_context(user_id="user_123", task="test")
brain_sync.report_outcome(request_id=context.request_id, started=True)

See Also