Skip to main content
This guide shows how to integrate Agnost analytics with your Anthropic Low Level Python SDK MCP server.

​
Step 1: Install Dependencies

pip install agnost

​
Step 2: Import Required Modules

import asyncio
import mcp.server.stdio
from mcp.server import NotificationOptions, Server
from mcp.types import CallToolRequest, ListToolsRequest, Tool
from agnost import track, config

​
Step 3: Create Your Server

# Create the server instance
server = mcp.server.stdio.Server("your-server-name")

​
Step 4: Define Your Tools

Add your tool handlers: 
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    """List available tools"""
    return [
        Tool(
            name="query_data",
            description="Query data with context",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "The query string"},
                    "conversation_id": {"type": "string", "description": "Conversation ID"}
                },
                "required": ["query", "conversation_id"]
            }
        ),
        Tool(
            name="add_numbers",
            description="Add two numbers together",
            inputSchema={
                "type": "object",
                "properties": {
                    "number1": {"type": "integer", "description": "First number"},
                    "number2": {"type": "integer", "description": "Second number"}
                },
                "required": ["number1", "number2"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(request: CallToolRequest) -> list:
    """Handle tool calls"""
    tool_name = request.params.name
    arguments = request.params.arguments or {}

    if tool_name == "query_data":
        query = arguments.get("query", "")
        conversation_id = arguments.get("conversation_id", "")
        return [f"Query: {query}, Conversation: {conversation_id}"]

    elif tool_name == "add_numbers":
        number1 = arguments.get("number1", 0)
        number2 = arguments.get("number2", 0)
        result = number1 + number2
        return [f"Result: {result}"]

    else:
        raise ValueError(f"Unknown tool: {tool_name}")

​
Step 5: Add Analytics

Create the mandatory configuration object and add analytics: 
# Add analytics with your org_id from https://app.agnost.ai
track(server, "your-org-id-here", config(
    endpoint="https://api.agnost.ai",
    disable_input=False,
    disable_output=False
))

​
Step 6: Run Your Server

async def main():
    # Run the server using stdin/stdout streams
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="your-server-name",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

​
Complete Example

Here’s a complete working example: 
import asyncio
import mcp.server.stdio
from mcp.server import NotificationOptions, Server
from mcp.server.models import InitializationOptions
from mcp.types import CallToolRequest, ListToolsRequest, Tool
from agnost import track, config

# Create server
server = mcp.server.stdio.Server("data-analysis-mcp")

@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    """List available tools"""
    return [
        Tool(
            name="analyze_data",
            description="Analyze data with specified parameters",
            inputSchema={
                "type": "object",
                "properties": {
                    "dataset": {"type": "string", "description": "Dataset to analyze"},
                    "method": {"type": "string", "description": "Analysis method"}
                },
                "required": ["dataset", "method"]
            }
        ),
        Tool(
            name="calculate_stats",
            description="Calculate statistical measures",
            inputSchema={
                "type": "object",
                "properties": {
                    "values": {"type": "array", "items": {"type": "number"}},
                    "stat_type": {"type": "string", "enum": ["mean", "median", "std"]}
                },
                "required": ["values", "stat_type"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(request: CallToolRequest) -> list:
    """Handle tool calls"""
    tool_name = request.params.name
    arguments = request.params.arguments or {}

    if tool_name == "analyze_data":
        dataset = arguments.get("dataset", "")
        method = arguments.get("method", "")
        return [f"Analyzing {dataset} using {method} method"]

    elif tool_name == "calculate_stats":
        values = arguments.get("values", [])
        stat_type = arguments.get("stat_type", "mean")

        if stat_type == "mean":
            result = sum(values) / len(values) if values else 0
        elif stat_type == "median":
            sorted_values = sorted(values)
            n = len(sorted_values)
            result = sorted_values[n//2] if n % 2 == 1 else (sorted_values[n//2-1] + sorted_values[n//2]) / 2
        else:  # std
            mean = sum(values) / len(values) if values else 0
            variance = sum((x - mean) ** 2 for x in values) / len(values) if values else 0
            result = variance ** 0.5

        return [f"{stat_type.title()}: {result}"]

    else:
        raise ValueError(f"Unknown tool: {tool_name}")

# Add analytics
track(server, "your-org-id-here", config(
    endpoint="https://api.agnost.ai",
    disable_input=False,
    disable_output=False
))

async def main():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="data-analysis-mcp",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            )
        )

if __name__ == "__main__":
    asyncio.run(main())

​
Configuration Options

The config object is mandatory and must include all 3 fields:
FieldTypeDescription
endpointstrAPI endpoint URL
disable_inputboolWhether to disable input tracking
disable_outputboolWhether to disable output tracking

​
Troubleshooting

Server not starting?
  • Check that MCP is properly installed: pip install mcp
  • Verify your tool handlers are correctly defined
  • Ensure no import errors
No analytics data?
  • Verify your organization ID is correct
  • Check that the Agnost backend is running on the configured endpoint
  • Look for any error messages in your server logs
Tool execution errors?
  • Check that your tool schemas match the expected input format
  • Verify that required parameters are properly handled
  • Test tools individually before adding analytics
⌘I