Agnost AI Analytics

---

name: agnost-ingestion description: USE when implementing data ingestion for Agnost AI analytics. Contains API reference, SDK guides for Python and TypeScript, and code examples for tracking AI conversations, MCP server events, and user interactions. license: MIT metadata: author: Agnost AI version: "1.0.0"

---

Agnost Data Ingestion

Comprehensive guide for ingesting data into Agnost AI for analytics, monitoring, and insights. Covers the Conversation SDK for tracking AI interactions and the MCP SDK for Model Context Protocol server analytics.

> Official docs: https://docs.agnost.ai > API Endpoint: https://api.agnost.ai > Dashboard: https://app.agnost.ai

IMPORTANT: How to Apply This Skill

Before implementing Agnost ingestion, follow this priority order:

1. Identify the use case: Conversation tracking (AI chatbots, agents) or MCP server analytics 2. Check SDK references in the references/ directory for detailed API 3. Use provided code examples as starting points 4. Cite references when explaining implementation details

---

Quick Reference

SDK Packages

| Use Case | Python | TypeScript/Node.js | Go | |----------|--------|-------------------|-----| | Conversation/AI Tracking | pip install agnost | npm install agnostai | N/A | | MCP Server Analytics | pip install agnost-mcp | npm install agnost | go get github.com/agnostai/agnost-go |

API Endpoints

| Endpoint | Method | Description | |----------|--------|-------------| | /api/v1/capture-session | POST | Create a new conversation/session | | /api/v1/capture-event | POST | Record an event within a session |

---

Conversation SDK (Recommended for AI Applications)

Use the Conversation SDK when building AI applications, chatbots, or agents that need to track user interactions, inputs, outputs, and performance metrics.

Python Installation & Setup

# Installation
pip install agnost
or
uv add agnost
Basic Setup
import agnost
Initialize with your org ID (from dashboard)
agnost.init("your-org-id")

TypeScript/Node.js Installation & Setup

// Installation
npm install agnostai
// or
pnpm add agnostai
// Basic Setup
import * as agnost from "agnostai";
// Initialize with your org ID (from dashboard)
agnost.init("your-org-id");

---

Core Methods

1. init(org_id, config?) - Initialize SDK

Must be called before any tracking methods.

#### Python

import agnost
Basic initialization
agnost.init("your-org-id")
With configuration
agnost.init(
    "your-org-id",
    endpoint="https://api.agnost.ai",  # Custom endpoint (optional)
    debug=True                          # Enable debug logging
)

#### TypeScript

import * as agnost from "agnostai";
// Basic initialization
agnost.init("your-org-id");
// With configuration
agnost.init("your-org-id", {
  endpoint: "https://api.agnost.ai",  // Custom endpoint (optional)
  debug: true                          // Enable debug logging
});

---

2. begin() + end() - Track Interactions (Recommended)

Use the begin/end pattern for automatic latency calculation and cleaner code.

#### Python

import agnost
agnost.init("your-org-id")
Start tracking an interaction
interaction = agnost.begin(
    user_id="user_123",
    agent_name="weather-agent",
    input="What's the weather in NYC?",
    conversation_id="conv_456",  # Optional: group related events
    properties={"model": "gpt-4"}  # Optional: custom metadata
)
... Your AI processing happens here ...
response = call_your_ai_model(interaction.input)
Complete the interaction (latency auto-calculated)
interaction.end(
    output=response,
    success=True  # Set False if the call failed
)

#### TypeScript

import * as agnost from "agnostai";
agnost.init("your-org-id");
// Start tracking an interaction
const interaction = agnost.begin({
  userId: "user_123",
  agentName: "weather-agent",
  input: "What's the weather in NYC?",
  conversationId: "conv_456",  // Optional: group related events
  properties: { model: "gpt-4" }  // Optional: custom metadata
});
// ... Your AI processing happens here ...
const response = await callYourAIModel(interaction.input);
// Complete the interaction (latency auto-calculated)
interaction.end(response);  // or interaction.end(response, true) for success

---

3. track() - Single-Call Tracking

Use when you have all data available at once (no need for begin/end).

#### Python

import agnost
agnost.init("your-org-id")
agnost.track(
    user_id="user_123",
    input="What's the weather?",
    output="The weather is sunny with 72°F.",
    agent_name="weather-agent",
    conversation_id="conv_456",  # Optional
    success=True,
    latency=150,  # milliseconds
    properties={"model": "gpt-4", "tokens": 42}
)

---

4. identify() - User Enrichment

Associate user metadata with a user ID for richer analytics.

#### Python

import agnost
agnost.init("your-org-id")
agnost.identify("user_123", {
    "name": "John Doe",
    "email": "john@example.com",
    "plan": "premium",
    "company": "Acme Inc"
})

#### TypeScript

import * as agnost from "agnostai";
agnost.init("your-org-id");
agnost.identify("user_123", {
  name: "John Doe",
  email: "john@example.com",
  plan: "premium",
  company: "Acme Inc"
});

---

5. flush() & shutdown() - Resource Management

#### Python

import agnost
Manually flush pending events
agnost.flush()
Clean shutdown (flushes and closes connections)
agnost.shutdown()

#### TypeScript

import * as agnost from "agnostai";
// Manually flush pending events
await agnost.flush();
// Clean shutdown (flushes and closes connections)
await agnost.shutdown();

---

Interaction Object Methods

When using begin(), you get an Interaction object with these methods:

| Method | Description | |--------|-------------| | set_input(text) / setInput(text) | Set/update the input text | | set_property(key, value) / setProperty(key, value) | Add a single custom property | | set_properties(dict) / setProperties(obj) | Add multiple custom properties | | end(output, success?, latency?) | Complete and send the event |

#### Example: Building Input Dynamically (Python)

interaction = agnost.begin(
    user_id="user_123",
    agent_name="my-agent"
)
Build input from multiple sources
interaction.set_input("Combined user query: " + user_input)
interaction.set_property("source", "chat-widget")
interaction.set_properties({"model": "gpt-4", "version": "v2"})
Process and complete
response = process_query(interaction.input)
interaction.end(output=response)

---

MCP Server Analytics

For tracking Model Context Protocol (MCP) servers, use the MCP SDK.

Python (FastMCP)

from mcp.server.fastmcp import FastMCP
from agnost_mcp import track, config
Create FastMCP server
mcp = FastMCP("my-mcp-server")
Add your tools
@mcp.tool()
def my_tool(param: str) -> str:
    return f"Result: {param}"
Enable tracking
track(mcp, "your-org-id", config(
    endpoint="https://api.agnost.ai",
    disable_input=False,   # Track input arguments
    disable_output=False   # Track output results
))
Run server
mcp.run()

TypeScript (MCP SDK)

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { trackMCP } from "agnost";
// Create MCP server
const server = new Server({
  name: "my-mcp-server",
  version: "1.0.0"
}, {
  capabilities: { tools: {} }
});
// Enable tracking
trackMCP(server, "your-org-id", {
  endpoint: "https://api.agnost.ai",
  disableInput: false,
  disableOutput: false
});
// Start server
const transport = new StdioServerTransport();
await server.connect(transport);

Go (mcp-go)

package main
import (
    "github.com/agnostai/agnost-go/agnost"
    "github.com/mark3labs/mcp-go/server"
)
func main() {
    s := server.NewMCPServer("my-server", "1.0.0")
    // Add tools...
    // Enable tracking
    agnost.Track(s, "your-org-id", &agnost.Config{
        DisableInput:  false,
        DisableOutput: false,
        BatchSize:     10,
        LogLevel:      "info",
    })
    server.ServeStdio(s)
}

---

API Reference (Direct HTTP)

For cases where you need direct API access without an SDK.

Create Session

curl -X POST https://api.agnost.ai/api/v1/capture-session \
  -H "Content-Type: application/json" \
  -H "X-Org-Id: your-org-id" \
  -d '{
    "session_id": "unique-session-id",
    "client_config": "my-app",
    "connection_type": "http",
    "ip": "",
    "user_data": {
      "user_id": "user_123",
      "email": "user@example.com"
    },
    "tools": ["tool1", "tool2"]
  }'

Capture Event

curl -X POST https://api.agnost.ai/api/v1/capture-event \
  -H "Content-Type: application/json" \
  -H "X-Org-Id: your-org-id" \
  -d '{
    "session_id": "unique-session-id",
    "primitive_type": "tool",
    "primitive_name": "weather_lookup",
    "latency": 150,
    "success": true,
    "args": "{\"city\": \"NYC\"}",
    "result": "{\"temp\": 72}",
    "metadata": {
      "model": "gpt-4",
      "tokens": "42"
    }
  }'

---

Data Structures

Session Request

{
  "session_id": "string (UUID or custom ID)",
  "client_config": "string (app identifier)",
  "connection_type": "string (http/stdio/sse)",
  "ip": "string (optional)",
  "user_data": {
    "user_id": "string",
    "...": "any additional user fields"
  },
  "tools": ["array", "of", "tool", "names"]
}

Event Request

{
  "session_id": "string (must match existing session)",
  "primitive_type": "string (tool/resource/prompt)",
  "primitive_name": "string (name of the primitive)",
  "latency": "integer (milliseconds)",
  "success": "boolean",
  "args": "string (JSON-encoded input)",
  "result": "string (JSON-encoded output)",
  "checkpoints": [
    {
      "name": "string",
      "timestamp": "integer (ms since start)",
      "metadata": {}
    }
  ],
  "metadata": {
    "key": "value pairs"
  }
}

---

Configuration Options

Python Conversation SDK

agnost.init(
    "your-org-id",
    endpoint="https://api.agnost.ai",  # API endpoint
    debug=False                         # Enable debug logging
)

TypeScript Conversation SDK

interface ConversationConfig {
  endpoint?: string;  // API endpoint (default: https://api.agnost.ai)
  debug?: boolean;    // Enable debug logging (default: false)
}
agnost.init("your-org-id", { endpoint: "...", debug: true });

Python MCP SDK (FastMCP)

from agnost_mcp import track, config
track(server, "your-org-id", config(
    endpoint="https://api.agnost.ai",
    disable_input=False,   # Don't track input arguments
    disable_output=False   # Don't track output results
))

TypeScript MCP SDK

import { trackMCP, createConfig } from "agnost";
const cfg = createConfig({
  endpoint: "https://api.agnost.ai",
  disableInput: false,
  disableOutput: false
});
trackMCP(server, "your-org-id", cfg);

Go MCP SDK

type Config struct {
    Endpoint         string        // default: "https://api.agnost.ai"
    DisableInput     bool          // default: false
    DisableOutput    bool          // default: false
    BatchSize        int           // default: 5
    MaxRetries       int           // default: 3
    RetryDelay       time.Duration // default: 1s
    RequestTimeout   time.Duration // default: 5s
    LogLevel         string        // "debug", "info", "warning", "error"
    Identify         IdentifyFunc  // optional user identification
}

---

Best Practices

1. Always Initialize Early

# At application startup
import agnost
agnost.init("your-org-id")

2. Use begin/end for Accurate Latency

# Automatically calculates processing time
interaction = agnost.begin(user_id="u1", agent_name="agent")
... processing ...
interaction.end(output=result)

3. Group Related Events with conversation_id

# All events for a single chat session
conversation_id = f"chat_{session_id}"
interaction = agnost.begin(
    user_id="u1",
    conversation_id=conversation_id,
    agent_name="chatbot"
)

4. Handle Errors Gracefully

interaction = agnost.begin(user_id="u1", agent_name="agent")
try:
    result = process_request()
    interaction.end(output=result, success=True)
except Exception as e:
    interaction.end(output=str(e), success=False)

5. Shutdown Cleanly

import atexit
import agnost
atexit.register(agnost.shutdown)

---

When to Apply

This skill activates when you encounter:

Data ingestion implementation for Agnost

AI conversation tracking setup

MCP server analytics integration

Event/session capture API usage

SDK initialization questions

Latency tracking requirements

User identification/enrichment

---

Additional Resources

Python SDK Reference: references/python-sdk.md

TypeScript SDK Reference: references/typescript-sdk.md

API Reference: references/api-reference.md

Dashboard: https://app.agnost.ai

Discord Community: https://discord.gg/agnost

1. Always Initialize Early

# At application startup
import agnost
agnost.init("your-org-id")

2. Use begin/end for Accurate Latency

# Automatically calculates processing time
interaction = agnost.begin(user_id="u1", agent_name="agent")
... processing ...
interaction.end(output=result)

3. Group Related Events with conversation_id

# All events for a single chat session
conversation_id = f"chat_{session_id}"
interaction = agnost.begin(
    user_id="u1",
    conversation_id=conversation_id,
    agent_name="chatbot"
)

4. Handle Errors Gracefully

interaction = agnost.begin(user_id="u1", agent_name="agent")
try:
    result = process_request()
    interaction.end(output=result, success=True)
except Exception as e:
    interaction.end(output=str(e), success=False)

5. Shutdown Cleanly

import atexit
import agnost
atexit.register(agnost.shutdown)

---