### Test Plugin Commands

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Examples of how to execute the 'hello' and 'calculate' commands from the plugin's CLI.

```bash
# Test hello command
python cli.py hello --name "World" --greeting "Hi"

# Test calculate command
python cli.py calculate --operation add --a 5 --b 3
```

--------------------------------

### Development Setup Commands

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/README.md

Commands for setting up the development environment, including installing dependencies, running linters, type checking, and tests.

```bash
# Install development dependencies
pip install -r requirements-dev.txt

# Run linting
flake8 smcp/ tests/

# Run type checking
mypy smcp/

# Run tests with coverage
python -m pytest tests/ --cov=smcp --cov-report=html
```

--------------------------------

### Development Setup Commands

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/README.md

Commands to set up the development environment. This covers installing dependencies, running linters, performing type checking, and executing tests with coverage reporting.

```bash
# Install development dependencies
pip install -r requirements-dev.txt

# Run linting
flake8 smcp/ tests/

# Run type checking
mypy smcp/

# Run tests with coverage
python -m pytest tests/ --cov=smcp --cov-report=html
```

--------------------------------

### Server Installation and Running

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/README.md

Instructions for cloning the repository, setting up a virtual environment, installing dependencies, and running the MCP server. Includes commands for default execution, localhost-only access, allowing external connections, and specifying custom ports or host bindings.

```bash
git clone https://github.com/actuallyrizzn/sanctum-letta-mcp.git
cd sanctum-letta-mcp
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install -r requirements.txt
python smcp/mcp_server.py

# Security Features
python smcp/mcp_server.py --host 127.0.0.1
python smcp/mcp_server.py --allow-external
python smcp/mcp_server.py --port 9000
python smcp/mcp_server.py --host 0.0.0.0 --port 8000
```

--------------------------------

### Python Command Setup with Complex Parameters

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Demonstrates how to set up a command-line interface command with various complex parameter types including strings, numbers, booleans, choices, and file paths using Python's argparse.

```python
def setup_advanced_command(subparsers):
    """Setup a command with complex parameters."""
    parser = subparsers.add_parser("advanced", help="Advanced command example")
    
    # String parameters
    parser.add_argument("--text", required=True, help="Text input")
    
    # Numeric parameters
    parser.add_argument("--number", type=int, default=10, help="Numeric input")
    parser.add_argument("--float", type=float, help="Float input")
    
    # Boolean parameters
    parser.add_argument("--verbose", action="store_true", help="Enable verbose output")
    parser.add_argument("--quiet", action="store_true", help="Suppress output")
    
    # Choice parameters
    parser.add_argument("--mode", choices=["fast", "safe", "debug"], 
                       default="safe", help="Operation mode")
    
    # File parameters
    parser.add_argument("--input-file", type=str, help="Input file path")
    parser.add_argument("--output-file", type=str, help="Output file path")
```

--------------------------------

### Production Deployment Steps

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Bash commands for packaging, deploying, and configuring the DevOps plugin for production environments.

```bash
# Package your plugin:
```bash
tar -czf your_plugin.tar.gz your_plugin/
```

# Deploy to server:
```bash
scp your_plugin.tar.gz user@server:/path/to/plugins/
ssh user@server "cd /path/to/plugins && tar -xzf your_plugin.tar.tar.gz"
```

# Update MCP server configuration:
```bash
export MCP_PLUGINS_DIR=/path/to/plugins
```
```

--------------------------------

### Complete MCP Client Integration Example

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/api-reference.md

Provides a comprehensive Python example using httpx to interact with the MCP server. It covers initialization, sending notifications, listing tools, and calling the health tool.

```python
import httpx
import json
import asyncio

async def mcp_client_example():
    """Complete MCP client integration example."""
    base_url = "http://localhost:8000"
    
    async with httpx.AsyncClient() as client:
        # Step 1: Initialize connection
        init_request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-03-26",
                "capabilities": {"tools": {}, "resources": {}, "prompts": {}},
                "clientInfo": {"name": "example-client", "version": "1.0.0"}
            }
        }
        
        response = await client.post(f"{base_url}/messages/", json=init_request)
        print("Initialization:", response.json())
        
        # Step 2: Send initialized notification
        await client.post(f"{base_url}/messages/", json={
            "jsonrpc": "2.0",
            "method": "initialized",
            "params": {}
        })
        
        # Step 3: List available tools
        tools_response = await client.post(f"{base_url}/messages/", json={
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list"
        })
        
        tools = tools_response.json()["result"]["tools"]
        print(f"Available tools: {len(tools)}")
        
        # Step 4: Call health tool
        health_response = await client.post(f"{base_url}/messages/", json={
            "jsonrpc": "2.0",
            "id": 3,
            "method": "tools/call",
            "params": {
                "name": "health",
                "arguments": {}
            }
        })
        
        health_result = health_response.json()["result"]
        print("Health check:", health_result)

# Run the example
asyncio.run(mcp_client_example())

```

--------------------------------

### DevOps Plugin CLI Commands

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Python code defining the command-line interface for the DevOps plugin, including setup for 'deploy' and 'rollback' commands with their respective arguments.

```python
#!/usr/bin/env python3
"""
DevOps Plugin

Deployment and infrastructure management for Sanctum Letta MCP.
"""

import argparse
import json
import sys
from typing import Dict, Any

def setup_deploy_command(subparsers):
    """Setup the deploy command."""
    parser = subparparsers.add_parser("deploy", help="Deploy an application")
    parser.add_argument("--app-name", required=True, help="Name of the application to deploy")
    parser.add_argument("--environment", default="production", help="Deployment environment")

def setup_rollback_command(subparsers):
    """Setup the rollback command."""
    parser = subparparsers.add_parser("rollback", help="Rollback an application")
    parser.add_argument("--app-name", required=True, help="Name of the application to rollback")
    parser.add_argument("--version", required=True, help="Version to rollback to")

def execute_deploy(app_name: str, environment: str) -> Dict[str, Any]:
    """Execute the deploy command."""
    # Implementation would handle actual deployment
    return {
        "status": "success",
        "message": f"Deployed {app_name} to {environment}",
        "data": {
            "app_name": app_name,
            "environment": environment,
            "deployment_id": "deploy-12345"
        }
    }

def execute_rollback(app_name: str, version: str) -> Dict[str, Any]:
    """Execute the rollback command."""
    # Implementation would handle actual rollback
    return {
        "status": "success",
        "message": f"Rolled back {app_name} to version {version}",
        "data": {
            "app_name": app_name,
            "version": version,
            "rollback_id": "rollback-67890"
        }
    }

```

--------------------------------

### Create Plugin Directory

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Bash commands to create the necessary directory structure for a new plugin.

```bash
mkdir -p smcp/plugins/my_first_plugin
cd smcp/plugins/my_first_plugin
```

--------------------------------

### Plugin Directory Structure Example

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/api-reference.md

Illustrates the standard file system layout for plugins within the MCP server. Each plugin resides in its own subdirectory, containing an __init__.py and a cli.py file.

```Bash
smcp/plugins/
├── botfather/
│   ├── __init__.py
│   └── cli.py
├── devops/
│   ├── __init__.py
│   └── cli.py
└── custom-plugin/
    ├── __init__.py
    └── cli.py
```

--------------------------------

### Symlink Creation Examples for Plugin Management

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/api-reference.md

Demonstrates the use of symbolic links (symlinks) to manage plugins. This allows for flexible deployment by pointing the MCP server's plugin directory to centralized or external plugin locations.

```Bash
# Create symlink to centralized plugin
ln -s /opt/sanctum/plugins/botfather smcp/plugins/botfather

# Create symlink to user's custom plugin
ln -s /home/user/custom-plugins/my-plugin smcp/plugins/my-plugin

# Create symlink to network-mounted plugin
ln -s /mnt/network/plugins/enterprise-plugin smcp/plugins/enterprise-plugin
```

--------------------------------

### Configuration via Environment Variables

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/README.md

Demonstrates how to configure the MCP server using environment variables for port, plugins directory, and host binding. Includes examples for setting variables and running the server with custom configurations.

```bash
# Default: localhost + Docker containers
python smcp/mcp_server.py

# Custom port
export MCP_PORT=9000
python smcp/mcp_server.py

# Localhost-only (more restrictive)
python smcp/mcp_server.py --host 127.0.0.1

# Custom plugins directory
export MCP_PLUGINS_DIR=/path/to/custom/plugins
python smcp/mcp_server.py
```

--------------------------------

### Python Asynchronous Operation with aiohttp

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Illustrates how to perform an asynchronous HTTP GET request using aiohttp and handle the response, including executing this async operation within a synchronous context.

```python
import asyncio
import aiohttp
from typing import Dict, Any

async def async_operation(url: str) -> Dict[str, Any]:
    """Perform an async operation."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.text()
            return {
                "status": "success",
                "url": url,
                "status_code": response.status,
                "data": data
            }

def execute_async_command(url: str) -> Dict[str, Any]:
    """Execute an async command."""
    try:
        result = asyncio.run(async_operation(url))
        return result
    except Exception as e:
        return {
            "status": "error",
            "error": str(e)
        }
```

--------------------------------

### Python Unit Tests for CLI Commands

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Demonstrates how to write unit tests for command-line interface commands using pytest and subprocess, verifying command execution, return codes, and JSON output.

```python
# tests/test_my_plugin.py
import pytest
import json
import subprocess
from pathlib import Path

def test_hello_command():
    """Test the hello command."""
    result = subprocess.run([
        "python", "cli.py", "hello", "--name", "Test", "--greeting", "Hi"
    ], capture_output=True, text=True)
    
    assert result.returncode == 0
    data = json.loads(result.stdout)
    assert data["status"] == "success"
    assert "Test" in data["message"]

def test_calculate_command():
    """Test the calculate command."""
    result = subprocess.run([
        "python", "cli.py", "calculate", "--operation", "add", "--a", "5", "--b", "3"
    ], capture_output=True, text=True)
    
    assert result.returncode == 0
    data = json.loads(result.stdout)
    assert data["status"] == "success"
    assert data["result"] == 8
```

--------------------------------

### Plugin Directory Structure

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Illustrates the standard directory layout for a Sanctum Letta MCP plugin, including essential files like cli.py and requirements.txt.

```plaintext
smcp/plugins/
├── your_plugin/
│   ├── __init__.py          # Plugin metadata (optional)
│   ├── cli.py              # Main plugin interface (required)
│   ├── requirements.txt    # Plugin dependencies (optional)
│   ├── README.md          # Plugin documentation (recommended)
│   └── tests/             # Plugin tests (recommended)
```

--------------------------------

### Make Plugin Executable

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Bash command to grant execute permissions to the plugin's main script.

```bash
chmod +x cli.py
```

--------------------------------

### MCP Implementation Example: Starlette

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/MCP-Reference-Architecture.md

A basic implementation snippet using Starlette to set up the MCP server. It demonstrates how to instantiate the SseServerTransport and integrate it into a Starlette application with defined routes.

```Python
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.responses import Response
from mcp.server.sse import SseServerTransport
from mcp.server.session import ServerSession
from mcp.server.models import InitializationOptions

# Create SSE transport
sse = SseServerTransport("/messages/")

# Example Starlette app setup (incomplete)
# routes = [
#     Route("/sse", endpoint=sse.connect_sse),
#     Route("/messages/", endpoint=sse.handle_post_message, methods=["POST"])
# ]
# app = Starlette(routes=routes)
```

--------------------------------

### Debugging CLI Commands

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Bash commands to test the DevOps plugin's CLI directly and check tool registration via HTTP requests.

```bash
# Test CLI directly:
```bash
python your_plugin/cli.py --help
python your_plugin/cli.py your-command --param value
```

# Verify tool registration:
```bash
curl -X POST http://localhost:8000/messages/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}'
```
```

--------------------------------

### Basic Plugin CLI Interface (Python)

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

A Python script demonstrating the core structure of a plugin's command-line interface (CLI). It includes argument parsing for 'hello' and 'calculate' commands and JSON output for results.

```python
#!/usr/bin/env python3
"""
My First Plugin

A sample plugin demonstrating basic plugin development.
"""

import argparse
import json
import sys
from typing import Dict, Any


def main():
    """Main entry point for the plugin CLI."""
    parser = argparse.ArgumentParser(
        description="My First Plugin - A sample plugin for Sanctum Letta MCP"
    )
    subparsers = parser.subparsers(dest="command", help="Available commands")
    
    # Add your commands here
    setup_hello_command(subparsers)
    setup_calculate_command(subparsers)
    
    args = parser.parse_args()
    
    if not args.command:
        parser.print_help()
        sys.exit(1)
    
    # Execute the appropriate command
    if args.command == "hello":
        result = execute_hello(args.name, args.greeting)
    elif args.command == "calculate":
        result = execute_calculate(args.operation, args.a, args.b)
    else:
        print(f"Unknown command: {args.command}")
        sys.exit(1)
    
    # Output JSON result
    print(json.dumps(result, indent=2))


def setup_hello_command(subparsers):
    """Setup the hello command."""
    parser = subparsers.add_parser("hello", help="Say hello to someone")
    parser.add_argument("--name", required=True, help="Name to greet")
    parser.add_argument("--greeting", default="Hello", help="Greeting message")


def setup_calculate_command(subparsers):
    """Setup the calculate command."""
    parser = subparsers.add_parser("calculate", help="Perform a calculation")
    parser.add_argument("--operation", required=True, 
                       choices=["add", "subtract", "multiply", "divide"],
                       help="Mathematical operation")
    parser.add_argument("--a", required=True, type=float, help="First number")
    parser.add_argument("--b", required=True, type=float, help="Second number")


def execute_hello(name: str, greeting: str) -> Dict[str, Any]:
    """Execute the hello command."""
    message = f"{greeting}, {name}!"
    return {
        "status": "success",
        "message": message,
        "data": {
            "name": name,
            "greeting": greeting
        }
    }


def execute_calculate(operation: str, a: float, b: float) -> Dict[str, Any]:
    """Execute the calculate command."""
    try:
        if operation == "add":
            result = a + b
        elif operation == "subtract":
            result = a - b
        elif operation == "multiply":
            result = a * b
        elif operation == "divide":
            if b == 0:
                raise ValueError("Division by zero")
            result = a / b
        else:
            raise ValueError(f"Unknown operation: {operation}")
        
        return {
            "status": "success",
            "result": result,
            "operation": operation,
            "operands": [a, b]
        }
    except Exception as e:
        return {
            "status": "error",
            "error": str(e),
            "operation": operation,
            "operands": [a, b]
        }


if __name__ == "__main__":
    main()

```

--------------------------------

### SSE Connection Example

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/api-reference.md

Demonstrates how to establish and listen to Server-Sent Events (SSE) from the MCP server using Python's httpx library. It shows how to connect to the SSE endpoint and process incoming data events.

```python
import httpx
import asyncio

async def sse_connection_example():
    """SSE connection example."""
    base_url = "http://localhost:8000"
    
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", f"{base_url}/sse") as response:
            print(f"SSE Status: {response.status_code}")
            print(f"Content-Type: {response.headers.get('content-type')}")
            
            # Read SSE events
            async for line in response.aiter_lines():
                if line.startswith("data:"):
                    data = line[5:].strip()
                    if data:
                        try:
                            event = json.loads(data)
                            print(f"SSE Event: {event}")
                        except json.JSONDecodeError:
                            print(f"Raw SSE data: {data}")

# Run the example
asyncio.run(sse_connection_example())

```

--------------------------------

### REST API Calls for MCP Server Management

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/Letta-MCP-Connection-Guide.md

Provides examples of using cURL to manage MCP servers via the Letta REST API, including adding, listing, and testing connections.

```bash
curl -X PUT "http://localhost:8080/v1/tools/mcp/servers" \
  -H "Content-Type: application/json" \
  -H "user_id: your_user_id" \
  -d '{
    "server_name": "my_mcp_server",
    "type": "sse",
    "server_url": "https://api.example.com/mcp/sse",
    "auth_header": "Authorization",
    "auth_token": "Bearer your_token"
  }'
```

```bash
curl -X GET "http://localhost:8080/v1/tools/mcp/servers" \
  -H "user_id: your_user_id"
```

```bash
curl -X POST "http://localhost:8080/v1/tools/mcp/servers/test" \
  -H "Content-Type: application/json" \
  -d '{
    "server_name": "my_mcp_server",
    "type": "sse",
    "server_url": "https://api.example.com/mcp/sse"
  }'
```

--------------------------------

### Official MCP Server Framework (Python)

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/Letta-MCP-Connection-Guide.md

Provides an example of using the official Python MCP framework to create an SSE server. This is the recommended approach for production environments as it ensures compatibility with Letta and handles the MCP protocol correctly. It demonstrates defining and running tools.

```bash
pip install mcp
```

```python
import mcp
import asyncio

# Define your tools
@mcp.tool()
async def example_tool(param1: str) -> str:
    """An example tool"""
    return f"Executed with parameter: {param1}"

# Run the server
if __name__ == "__main__":
    mcp.run(transport="sse", port=5000)
```

--------------------------------

### Log Format Example

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/api-reference.md

Illustrates the structured log format used by the Sanctum Letta MCP server. It includes timestamp, module, log level, and the log message.

```APIDOC
2025-07-11 15:21:07,215 - __main__ - INFO - Starting Sanctum Letta MCP Server...
2025-07-11 15:21:07,354 - __main__ - INFO - Registered tool: botfather.click-button
```

--------------------------------

### MCP Plugin Directory Structure Example

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/project-plan.md

Illustrates the expected file and directory layout for MCP plugins, emphasizing the 'cli.py' entrypoint required for autodiscovery.

```text
mcp/
  plugins/
    botfather/
      cli.py
      utils.py
    devops/
      cli.py
      deploy.py
```

--------------------------------

### Test Plugin Integration with MCP Server

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

This Python script tests the integration of a plugin with the MCP server using httpx. It sends initialization, lists tools, and calls a specific tool, asserting the responses.

```python
import httpx
import json

async def test_plugin_integration():
    """Test plugin integration with MCP server."""
    base_url = "http://localhost:8000"
    
    async with httpx.AsyncClient() as client:
        # Initialize connection
        init_request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-03-26",
                "capabilities": {"tools": {}},
                "clientInfo": {"name": "test-client", "version": "1.0.0"}
            }
        }
        
        response = await client.post(f"{base_url}/messages/", json=init_request)
        assert response.status_code == 200
        
        # List tools to find your plugin
        tools_request = {
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list"
        }
        
        response = await client.post(f"{base_url}/messages/", json=tools_request)
        tools = response.json()["result"]["tools"]
        
        # Find your plugin's tools
        your_tools = [t for t in tools if t["name"].startswith("my_first_plugin.")]
        assert len(your_tools) > 0
        
        # Test calling your tool
        tool_name = your_tools[0]["name"]
        call_request = {
            "jsonrpc": "2.0",
            "id": 3,
            "method": "tools/call",
            "params": {
                "name": tool_name,
                "arguments": {"name": "Test", "greeting": "Hi"}
            }
        }
        
        response = await client.post(f"{base_url}/messages/", json=call_request)
        result = response.json()["result"]
        assert "content" in result
```

--------------------------------

### Starlette SSE Server Setup

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/MCP-Reference-Architecture.md

Defines an asynchronous handler for Server-Sent Events (SSE) using Starlette. It sets up a server session with initialization options and runs the session. The code also configures routes for SSE and POST message handling.

```Python
from starlette.applications import Starlette
from starlette.routing import Route, Mount

# Assuming sse, InitializationOptions, ServerCapabilities, ToolsCapability, ResourcesCapability, ServerSession, Response are imported from relevant libraries

async def handle_sse(request):
    async with sse.connect_sse(
        request.scope, request.receive, request._send
    ) as streams:
        # Create server session
        init_options = InitializationOptions(
            server_name="my-mcp-server",
            server_version="1.0.0",
            capabilities=ServerCapabilities(
                tools=ToolsCapability(),
                resources=ResourcesCapability()
            )
        )
        
        session = ServerSession(
            read_stream=streams[0],
            write_stream=streams[1],
            init_options=init_options
        )
        
        # Run the session
        await session.run()
    
    return Response()

# Create routes
routes = [
    Route("/sse", endpoint=handle_sse, methods=["GET"]),
    Mount("/messages/", app=sse.handle_post_message),
]

# Create and run app
app = Starlette(routes=routes)
```

--------------------------------

### Python Flask SSE Server and MCP API

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/Letta-MCP-Connection-Guide.md

A Python Flask application demonstrating how to set up a Server-Sent Events (SSE) endpoint for client connections and an API endpoint to handle JSON-RPC 2.0 messages. This example includes basic logic for initializing the server, listing tools, and executing tools, but notes that it's a simplified version for demonstration and may require adjustments for full production compatibility with specific MCP libraries.

```python
from flask import Flask, Response, request
import json
import uuid
import asyncio
import threading
from queue import Queue

app = Flask(__name__)

# Store active connections and message queues
connections = {}
message_queues = {}

@app.route('/mcp/sse')
def mcp_sse():
    def generate():
        # Generate unique connection ID
        conn_id = str(uuid.uuid4())
        connections[conn_id] = True
        message_queues[conn_id] = Queue()
        
        try:
            # Wait for client to send messages
            while connections.get(conn_id):
                # In a real implementation, you would handle bidirectional communication
                # This is a simplified example - you need to implement proper message handling
                pass
                
        except GeneratorExit:
            # Client disconnected
            connections.pop(conn_id, None)
            message_queues.pop(conn_id, None)
    
    return Response(
        generate(),
        mimetype='text/event-stream',
        headers={
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Headers': 'Content-Type, Authorization'
        }
    )

@app.route('/mcp/message', methods=['POST'])
def handle_mcp_message():
    message = request.json
    
    # Validate JSON-RPC 2.0 message
    if not isinstance(message, dict) or 'jsonrpc' not in message or message['jsonrpc'] != '2.0':
        return json.dumps({
            "jsonrpc": "2.0",
            "id": message.get('id'),
            "error": {
                "code": -32600,
                "message": "Invalid Request"
            }
        })
    
    # Handle different MCP message types
    if message.get('method') == 'initialize':
        response = {
            "jsonrpc": "2.0",
            "id": message.get('id'),
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {
                    "tools": {}
                },
                "serverInfo": {
                    "name": "example-mcp-server",
                    "version": "1.0.0"
                }
            }
        }
    elif message.get('method') == 'tools/list':
        response = {
            "jsonrpc": "2.0",
            "id": message.get('id'),
            "result": {
                "tools": [
                    {
                        "name": "example_tool",
                        "description": "An example tool",
                        "inputSchema": {
                            "type": "object",
                            "properties": {
                                "param1": {
                                    "type": "string",
                                    "description": "First parameter"
                                }
                            },
                            "required": ["param1"]
                        }
                    }
                ]
            }
        }
    elif message.get('method') == 'tools/call':
        # Execute the tool
        tool_name = message['params']['name']
        arguments = message['params']['arguments']
        
        # Your tool execution logic here
        result = execute_tool(tool_name, arguments)
        
        response = {
            "jsonrpc": "2.0",
            "id": message.get('id'),
            "result": {
                "content": [
                    {
                        "type": "text",
                        "text": result
                    }
                ]
            }
        }
    else:
        response = {
            "jsonrpc": "2.0",
            "id": message.get('id'),
            "error": {
                "code": -32601,
                "message": "Method not found"
            }
        }
    
    return json.dumps(response)

def execute_tool(tool_name, arguments):
    # Implement your tool execution logic here
    if tool_name == "example_tool":
        return f"Executed {tool_name} with arguments: {arguments}"
    return "Tool not found"

if __name__ == '__main__':
    app.run(debug=True, port=5000)

```

--------------------------------

### Python Function Execution with Error Handling

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

Provides a Python function to execute another function, capturing and returning specific exceptions (ValueError, FileNotFoundError) and general exceptions in a structured JSON format.

```python
def execute_with_error_handling(func, *args, **kwargs):
    """Execute a function with proper error handling."""
    try:
        result = func(*args, **kwargs)
        return {
            "status": "success",
            "result": result
        }
    except ValueError as e:
        return {
            "status": "error",
            "error": f"Invalid input: {str(e)}",
            "error_type": "validation_error"
        }
    except FileNotFoundError as e:
        return {
            "status": "error",
            "error": f"File not found: {str(e)}",
            "error_type": "file_error"
        }
    except Exception as e:
        return {
            "status": "error",
            "error": f"Unexpected error: {str(e)}",
            "error_type": "unknown_error"
        }
```

--------------------------------

### BotFather Plugin for Telegram Integration

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/plugin-development-guide.md

This Python code defines commands for the BotFather plugin, which integrates with the Telegram Bot API. It includes functions to set up argument parsers for 'click-button' and 'send-message' commands, and functions to execute these commands.

```python
#!/usr/bin/env python3
"""
BotFather Plugin

Telegram Bot API integration for Sanctum Letta MCP.
"""

import argparse
import json
import sys
import requests
from typing import Dict, Any

def setup_click_button_command(subparsers):
    """Setup the click-button command."""
    parser = subparsers.add_parser("click-button", help="Click a button in a message")
    parser.add_argument("--button-text", required=True, help="Text of the button to click")
    parser.add_argument("--msg-id", required=True, type=int, help="Message ID containing the button")

def setup_send_message_command(subparsers):
    """Setup the send-message command."""
    parser = subparsers.add_parser("send-message", help="Send a message")
    parser.add_argument("--message", required=True, help="Message to send")
    parser.add_argument("--chat-id", required=True, help="Chat ID to send to")

def execute_click_button(button_text: str, msg_id: int) -> Dict[str, Any]:
    """Execute the click-button command."""
    # Implementation would interact with Telegram Bot API
    return {
        "status": "success",
        "message": f"Clicked button '{button_text}' on message {msg_id}",
        "data": {
            "button_text": button_text,
            "message_id": msg_id
        }
    }

def execute_send_message(message: str, chat_id: str) -> Dict[str, Any]:
    """Execute the send-message command."""
    # Implementation would send message via Telegram Bot API
    return {
        "status": "success",
        "message": f"Sent message to chat {chat_id}",
        "data": {
            "message": message,
            "chat_id": chat_id
        }
    }
```

--------------------------------

### MCP Legacy HTTP/SSE API - GET/POST /help Endpoint

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/project-plan.md

Describes the /help endpoint, used to retrieve information about available plugins, their commands, arguments, and usage examples.

```APIDOC
GET /help or POST /help
  Description: Returns list of available plugins, their commands, required/optional arguments, and usage examples.
  Response Example:
    {
      "plugins": {
        "botfather": {
          "send-message": {
            "description": "Send a message to BotFather",
            "args": ["--msg"]
          },
          "get-replies": {
            "description": "Get replies from BotFather",
            "args": ["--limit"]
          },
          "click-button": {
            "description": "Click a button in BotFather's message",
            "args": ["--button-text", "--row", "--col", "--msg-id"]
          }
        },
        "devops": {
          "deploy": { ... }
        }
      }
    }
```

--------------------------------

### Node.js Express SSE Server for MCP

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/Letta-MCP-Connection-Guide.md

Sets up a basic Server-Sent Events (SSE) server using Node.js and Express. It handles client connections and processes incoming JSON-RPC 2.0 messages for methods like 'initialize', 'tools/list', and 'tools/call'. Note: This is a simplified example and may require adjustments for full compatibility with Letta's specific bidirectional protocol.

```javascript
const express = require('express');
const app = express();

app.use(express.json());

// Store active connections
const connections = new Map();

app.get('/mcp/sse', (req, res) => {
    // Set SSE headers
    res.writeHead(200, {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Headers': 'Content-Type, Authorization'
    });

    const connectionId = Date.now().toString();
    connections.set(connectionId, res);

    // Handle client disconnect
    req.on('close', () => {
        connections.delete(connectionId);
    });
});

app.post('/mcp/message', (req, res) => {
    const message = req.body;
    let response;

    // Validate JSON-RPC 2.0 message
    if (!message || message.jsonrpc !== '2.0') {
        response = {
            jsonrpc: "2.0",
            id: message?.id,
            error: {
                code: -32600,
                message: "Invalid Request"
            }
        };
        return res.json(response);
    }

    switch (message.method) {
        case 'initialize':
            response = {
                jsonrpc: "2.0",
                id: message.id,
                result: {
                    protocolVersion: "2024-11-05",
                    capabilities: {
                        tools: {}
                    },
                    serverInfo: {
                        name: "example-mcp-server",
                        version: "1.0.0"
                    }
                }
            };
            break;

        case 'tools/list':
            response = {
                jsonrpc: "2.0",
                id: message.id,
                result: {
                    tools: [
                        {
                            name: "example_tool",
                            description: "An example tool",
                            inputSchema: {
                                type: "object",
                                properties: {
                                    param1: {
                                        type: "string",
                                        description: "First parameter"
                                    }
                                },
                                required: ["param1"]
                            }
                        }
                    ]
                }
            };
            break;

        case 'tools/call':
            const toolName = message.params.name;
            const arguments = message.params.arguments;
            
            // Execute tool logic here
            const result = executeTool(toolName, arguments);
            
            response = {
                jsonrpc: "2.0",
                id: message.id,
                result: {
                    content: [
                        {
                            type: "text",
                            text: result
                        }
                    ]
                }
            };
            break;

        default:
            response = {
                jsonrpc: "2.0",
                id: message.id,
                error: {
                    code: -32601,
                    message: "Method not found"
                }
            };
    }

    res.json(response);
});

function executeTool(toolName, arguments) {
    if (toolName === "example_tool") {
        return `Executed ${toolName} with arguments: ${JSON.stringify(arguments)}`;
    }
    return "Tool not found";
}

app.listen(5000, () => {
    console.log('MCP Server running on port 5000');
});
```

--------------------------------

### MCP Server Health Check

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/README.md

Example cURL command to perform a health check on the MCP server by sending a JSON-RPC 2.0 'tools/call' request for the 'health' tool.

```APIDOC
MCP Server Endpoints:

POST /messages/
  Description: Handles JSON-RPC 2.0 messages for communication with the MCP server.
  Methods:
    initialize:
      Description: Initializes the client connection and exchanges capabilities.
      Parameters:
        protocolVersion (string, required): The version of the MCP protocol.
        capabilities (object, required): Client's supported capabilities (tools, resources, prompts).
        clientInfo (object, required): Information about the client (name, version).
      Returns: Server capabilities and status.
    
    tools/list:
      Description: Requests a list of available tools on the server.
      Returns:
        tools (array): An array of available tool objects.
    
    tools/call:
      Description: Executes a specific tool on the server.
      Parameters:
        name (string, required): The name of the tool to execute.
        arguments (object, optional): Arguments to pass to the tool.
      Returns:
        Result of the tool execution.

Example Health Check using cURL:

curl -X POST http://localhost:8000/messages/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"health","arguments":{}}}'
```

--------------------------------

### Server Configuration with FastMCP

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/api-reference.md

Illustrates how to initialize and configure the FastMCP server instance in Python, specifying parameters like name, instructions, SSE path, message path, host, and port.

```python
# Server configuration
server = FastMCP(
    name="sanctum-letta-mcp",
    instructions="A plugin-based MCP server for Sanctum Letta operations",
    sse_path="/sse",
    message_path="/messages/",
    host=os.getenv("MCP_HOST", "0.0.0.0"),
    port=int(os.getenv("MCP_PORT", "8000"))
)
```

--------------------------------

### Plugin Development Structure and Deployment

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/README.md

Outlines the standard directory structure for MCP plugins and explains how to use symbolic links for flexible plugin management. Covers centralized plugin repositories and overriding the default plugin directory via environment variables.

```bash
# Plugin Structure
# plugins/
# ├── your_plugin/
# │   ├── __init__.py
# │   ├── cli.py          # Main plugin interface
# │   └── README.md       # Plugin documentation

# Plugin Deployment with Symlinks
# Central plugin repository
# /opt/sanctum/plugins/
# ├── botfather/
# ├── devops/
# └── custom-plugin/

# MCP server plugin directory with symlinks
# smcp/plugins/
# ├── botfather -> /opt/sanctum/plugins/botfather
# ├── devops -> /opt/sanctum/plugins/devops
# └── custom-plugin -> /opt/sanctum/plugins/custom-plugin

# Environment Variable Override
export MCP_PLUGINS_DIR=/opt/sanctum/plugins
python smcp/mcp_server.py
```

--------------------------------

### Python MCP Server Framework

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/Letta-MCP-Connection-Guide.md

Demonstrates how to use the official MCP server framework in Python for robust compatibility with Letta. It shows how to define tools using decorators and run the server, typically over SSE. This is the recommended approach for production environments.

```bash
pip install mcp
```

```python
import mcp
import asyncio

# Define your tools
@mcp.tool()
async def example_tool(param1: str) -> str:
    """An example tool"""
    return f"Executed with parameter: {param1}"

# Run the server
if __name__ == "__main__":
    mcp.run(transport="sse", port=5000)
```

--------------------------------

### Python Configuration for SSE Server

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/Letta-MCP-Connection-Guide.md

Illustrates how to configure an SSE server connection using the SSEServerConfig class with different authentication methods.

```python
SSEServerConfig(
    server_name="my_server",
    server_url="https://api.example.com/mcp/sse",
    auth_header="Authorization",
    auth_token="Bearer your_token_here"
)
```

```python
SSEServerConfig(
    server_name="my_server",
    server_url="https://api.example.com/mcp/sse",
    custom_headers={
        "X-API-Key": "your_api_key_here",
        "X-Custom-Header": "custom_value"
    }
)
```

```python
SSEServerConfig(
    server_name="my_server",
    server_url="https://api.example.com/mcp/sse"
)
```

--------------------------------

### MCP Protocol Client Integration

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/README.md

Demonstrates how to integrate with the MCP protocol using Python's httpx library. It covers initializing a connection, listing available tools, and calling a tool.

```python
import httpx
import json

async def connect_to_mcp():
    base_url = "http://localhost:8000"
    
    # Initialize connection
    init_request = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-03-26",
            "capabilities": {"tools": {}, "resources": {}, "prompts": {}},
            "clientInfo": {"name": "my-client", "version": "1.0.0"}
        }
    }
    
    async with httpx.AsyncClient() as client:
        response = await client.post(f"{base_url}/messages/", json=init_request)
        data = response.json()
        
        # List available tools
        tools_request = {
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list"
        }
        
        response = await client.post(f"{base_url}/messages/", json=tools_request)
        tools = response.json()["result"]["tools"]
        
        # Call a tool
        call_request = {
            "jsonrpc": "2.0",
            "id": 3,
            "method": "tools/call",
            "params": {
                "name": "health",
                "arguments": {}
            }
        }
        
        response = await client.post(f"{base_url}/messages/", json=call_request)
        result = response.json()["result"]
        
        return result
```

--------------------------------

### Legacy MCP Server Configuration File

Source: https://github.com/actuallyrizzn/sanctum-letta-mcp/blob/master/docs/Letta-MCP-Connection-Guide.md

Shows the format for configuring MCP servers using a JSON file located at `~/.letta/mcp_config.json`.

```json
{
  "mcpServers": {
    "my_mcp_server": {
      "transport": "sse",
      "url": "https://api.example.com/mcp/sse",
      "headers": {
        "Authorization": "Bearer your_token_here"
      }
    }
  }
}
```