log-sink-mcp

AI-powered log analysis MCP server with HTTP ingestion and SQLite storage.

Overview

log-sink-mcp provides a complete log analysis solution for AI assistants like Claude. It combines:

• HTTP Server: REST endpoint for log ingestion from any application
• MCP Server: 5 application-focused tools for searching/filtering, tracing, and inspecting logs
• CLI Analysis Commands: Local shell commands for deeper error and agent-issue analysis
• Pi Extension: Ships a Pi extension that exports the agent's own session telemetry into the sink
• SQLite Database: Fast local storage with FTS5 full-text search plus cached local embeddings
• Real-time Analysis: Query logs, trace distributed requests, analyze error patterns

Perfect for debugging, monitoring, and understanding application behavior through AI-powered analysis.

Features

• ✅ HTTP Log Ingestion: POST logs from any application via REST API
• ✅ 5 MCP Tools: Search/filter, trace, agent-issue analysis, and inspect stats/services
• ✅ Agent Telemetry: Pi extension records prompts, turns, tool calls, token usage, and cost
• ✅ Hybrid Search: FTS5-powered search plus optional local semantic retrieval across messages and errors
• ✅ Distributed Tracing: Timeline view for trace IDs and span relationships
• ✅ Error Analysis: Pattern detection and error categorization
• ✅ Structured Logging: JSON metadata, trace IDs, span IDs
• ✅ Auto-Migration: Drizzle ORM with automatic schema migrations
• ✅ Type-Safe: Full TypeScript coverage with Drizzle schema inference

Architecture

┌─────────────┐        HTTP POST         ┌─────────────────┐
│ Application │ ───────────────────────> │  HTTP Server    │
│  (Pino,     │     /logs endpoint       │  (Hono)         │
│  Winston)   │                          │  Port 3100      │
└─────────────┘                          └────────┬────────┘
                                                  │
                                                  ▼
                                         ┌─────────────────┐
                                         │  LogStorage     │
                                         │  Service        │
                                         └────────┬────────┘
                                                  │
                                                  ▼
┌──────────────┐       MCP Protocol      ┌─────────────────┐
│   Claude /   │ <─────────────────────> │   MCP Server    │
│   AI Agent   │    5 App Log Tools      │   (stdio)       │
└──────────────┘                         └────────┬────────┘
                                                  │
                                                  ▼
                                         ┌─────────────────┐
                                         │ SQLite + FTS5 + │
                                         │  local vectors   │
                                         │  ./logs/        │
                                         │  session.db     │
                                         └─────────────────┘

Singleton HTTP Server Management

log-sink-mcp implements automatic singleton coordination to prevent port conflicts when multiple MCP servers run concurrently.

How It Works

1. Service Discovery: When an MCP server starts, it checks .agiflow/registry/log-sink-mcp-http_{env}.json for an existing HTTP server
2. Health Check: Pings /health endpoint to verify the registered server is alive
3. Reuse or Start: If healthy, reuses existing server; otherwise starts a new one
4. PID Tracking: Process IDs stored in .pids/ for lifecycle management
5. Auto-Cleanup: Registry and PID files removed on graceful shutdown

Service Registry Schema

{
  "name": "log-sink-mcp-http",
  "port": 3100,
  "host": "localhost",
  "environment": "development",
  "pid": 12345,
  "status": "running",
  "startedAt": "2026-01-15T10:00:00.000Z",
  "metadata": {
    "healthCheckUrl": "http://localhost:3100/health",
    "dbPath": "./logs/session.db"
  }
}

Management Commands

# Check status of HTTP server
bun run src/cli.ts status

# Stop HTTP server
bun run src/cli.ts stop

# Start with auto-coordination
bun run src/cli.ts start

Installation

# From monorepo root
pnpm install

# Optional: install local embedding support for semantic search
pnpm add ruvector-onnx-embeddings-wasm

# Or install globally
pnpm add -g log-sink-mcp

Semantic embeddings are disabled by default. Set LOG_SINK_ENABLE_EMBEDDINGS=true to enable local model loading and semantic indexing. If ruvector-onnx-embeddings-wasm is not installed, log-sink-mcp still starts and falls back to FTS5 search only.

Quick Start

1. Start the Servers

# Start both HTTP and MCP servers (recommended)
bun run src/cli.ts start

# Or start only HTTP server
bun run src/cli.ts start --http-only

# Or start only MCP server (stdio transport)
bun run src/cli.ts start --mcp-only

# With custom configuration
bun run src/cli.ts start --port 3100 --db-path ./logs/session.db

The start command uses singleton pattern - multiple MCP servers will share a single HTTP server automatically.

Local vs Global Log Sink

By default, log-sink resolves to a repo-local instance. Add a .logsink.yaml file when selected MCP/tooling packages should send telemetry to a shared global instance instead:

global:
  port: 3100
  services:
    - "@agimon-ai/log-sink-mcp"
    - "@agimon-ai/mcp-proxy"
    - "@agimon-ai/browse-tool"
    - "@agimon-ai/workflow-mcp"

Only exact package/service names listed in global.services use the global instance. Unlisted names stay local. If global.port is omitted, log-sink asks the port registry to allocate and register an available global port dynamically.

Config lookup uses LOG_SINK_CONFIG or --config <path> first, then the nearest .logsink.yaml from the current directory upward, then ~/.logsink.yaml. The global instance stores its registry identity and default database under ~/.log-sink-mcp/global/.

Useful overrides:

# Show currently registered ports
log-sink-mcp port --all --no-health-check
log-sink-mcp port --instance global
log-sink-mcp port --global

# Force one command to manage/query the local instance
log-sink-mcp status --instance local

# Query logs from the global instance
log-sink-mcp logs query --global --limit 20

# Send telemetry to an explicit running log-sink endpoint
LOG_SINK_ENDPOINT=http://127.0.0.1:3100 node ./dist/cli.mjs mcp-serve

2. Configure MCP Server

Add to your mcp-config.yaml:

mcpServers:
  log-sink:
    type: stdio
    command: bun
    args:
      - ./packages/log-sink-mcp/src/cli.ts
      - mcp-serve
      - --db-path
      - ./logs/session.db
    disabled: false

3. Send Logs

# Send a test log
curl -X POST http://localhost:3100/logs \
  -H "Content-Type: application/json" \
  -d '{
    "logs": [{
      "level": "info",
      "message": "Application started",
      "service": "my-app",
      "metadata": {"version": "1.0.0"}
    }]
  }'

4. Query with AI

Ask Claude:

"Query all error logs from the database"

"Show me the trace timeline for trace ID abc123"

"Analyze error patterns in the logs"

5. Claude Code Hook Context

Use the hook command to surface fresh errors back into Claude Code after each tool use:

{
  "hooks": {
    "PostToolUse": [
      {
        "command": "log-sink-mcp",
        "args": ["claude-hook"]
      }
    ],
    "PostToolUseFailure": [
      {
        "command": "log-sink-mcp",
        "args": ["claude-hook"]
      }
    ]
  }
}

The hook reads the Claude hook payload from stdin, checks the current log database, and emits additionalContext only when there are new error or fatal entries. After it notifies Claude, it advances its checkpoint so the same errors are not reported again.

HTTP API Reference

POST /logs

Ingest log entries in batch.

Request Body:

{
  logs: Array<{
    // Required fields
    level: 'debug' | 'info' | 'warn' | 'error';
    message: string;
    service: string;

    // Optional fields
    timestamp?: string;        // ISO 8601, defaults to now
    traceId?: string;          // 32-char hex for distributed tracing
    spanId?: string;           // 16-char hex for span identification
    parentSpanId?: string;     // 16-char hex for parent span
    hostname?: string;
    pid?: number;
    metadata?: Record<string, any>;
    errorType?: string;
    errorMessage?: string;
    errorStack?: string;
  }>
}

Response (201 Created):

{
  "success": true,
  "count": 5,
  "message": "Successfully inserted 5 logs"
}

Response (400 Bad Request):

{
  "success": false,
  "error": "Validation failed",
  "details": [...]
}

GET /health

Health check endpoint.

Response (200 OK):

{
  "status": "healthy",
  "service": "log-sink-mcp",
  "timestamp": "2026-01-15T10:00:00.000Z"
}

MCP Tools Reference

1. search_logs

Search and filter log entries. Without query/searchQuery, it behaves as a structured filter tool. With query or searchQuery, it searches across log messages and error messages. By default text search combines:

• FTS5 exact / keyword matching
• Local semantic similarity from cached embeddings

Parameters:

• query? / searchQuery?: string - Optional text to search for. Common punctuation such as hyphens is escaped safely; quoted phrases, prefix matching with *, and boolean operators AND/OR/NOT are supported.
• mode?: string - hybrid (default), fts, or semantic
• service?: string or string[] - Filter by service name
• level?: string or string[] - Filter by log level
• traceId?: string - Filter by trace ID
• spanId?: string - Filter by span ID
• errorType?: string - Filter by error type
• sessionId? / parentSessionId?: string - Filter by agent session ID or its parent
• agentName?: string - Filter by agent name
• workflowRunId? / workflowRunKey? / workflowId?: string - Filter by workflow run ID, stable run key, or workflow file ID
• workflowName? / workflowWorkspace?: string - Filter by workflow name or workspace
• jobName? / jobId?: string - Filter by workflow job name or job execution ID
• stepName? / phase?: string - Filter by workflow step name or phase
• signalType?: string - log or span
• startTime?: string - ISO 8601 start time filter
• endTime?: string - ISO 8601 end time filter
• limit?: number - Maximum results (default: 100)

Example:

Find error logs from user-service that mention "database timeout"

Returns: Matching logs with count and filters. Text searches also include search relevance metadata.

---

2. analyze_agent_issues

Detect and aggregate agent-run problems for reviewing agent performance and improving the harness: failed tool calls, API errors, refusals, retries exhausted, rejected tools, and error logs. Unlike error-severity analysis, this understands agent event semantics, so it also catches warning-level returned tool errors and info-level tool results carrying success=false.

Parameters:

• sessionId? / parentSessionId?: string - Scope to one agent launch
• agentName?: string - Scope to one agent identity
• service?: string or string[] - Scope by agent/service name (for example claude-code, codex)
• workflowRunId? / workflowRunKey? / workflowName? / workflowWorkspace?: string - Scope by workflow
• jobName? / jobId? / stepName? / phase?: string - Scope within a run
• startTime? / endTime?: string - ISO 8601 analysis window
• limit?: number - Maximum issue samples returned (default: 100)
• scanLimit?: number - Maximum matching records scanned (default: 50000); the response reports when this truncates the window

Example:

Why did the last agent session keep failing its edits?

Returns: Counts by category, tool, and error type, plus sample issues carrying the tool, input, and error context needed to diagnose them.

---

3. get_trace_timeline

Get chronological timeline for a distributed trace.

Parameters:

• traceId: string (required) - Trace ID to retrieve

Example:

Show me the trace timeline for trace ID a1b2c3d4e5f6789012345678901234ab

Returns: Ordered sequence of log entries for the trace, showing service call flow.

---

4. get_log_stats

Get aggregate statistics, grouped along whichever dimension you ask for.

Parameters:

• groupBy?: string - both (default), level, service, agent, workflow, job, or signal
• signalType?: string - log (default), span, or all
• startTime? / endTime?: string - ISO 8601 window

Example:

Show me log statistics

Returns: Counts per group, plus a total.

---

5. get_services

List all unique services that have logged.

Parameters: None

Example:

What services are logging to the database?

Returns: Array of service names with log counts.

Pi extension

The package ships a Pi extension, declared in the pi.extensions field of package.json and loaded when the package is installed as a Pi package. It registers no commands and no tools. It subscribes to Pi's session events and exports them as OTLP logs under the service name pi, so the agent's own behaviour lands in the same database as your application's.

Pi event	Record	Level
session_start	pi.session.started	info
before_agent_start	pi.user_prompt	info
agent_start	pi.agent.started	info
turn_start	pi.turn.started	debug
turn_end	pi.turn.finished	info
tool_execution_start	pi.tool_call	debug
tool_execution_end	pi.tool_result	info
after_provider_response	pi.api_response, or pi.api_error on status 400 and above	debug / error
model_select	pi.model.selected	info
agent_end	pi.agent.finished	info
agent_settled	pi.agent.settled	debug
session_shutdown	pi.session.finished	info

Every record carries the session ID, cwd, mode, thinking level, and selected model. pi.turn.finished adds the standard gen_ai.usage.* attributes: input, output, cache read, cache write, reasoning, total tokens, and cost. pi.tool_result adds the tool name and its duration in milliseconds.

Two things are deliberate about what it does not do:

• Prompts and tool payloads are omitted by default. Set AGENT_OTEL_REDACT=0 to include user.prompt, tool.input, and tool.result.
• Telemetry is dropped rather than written to disk when no sink is reachable. Endpoint discovery health-checks a running server; if there is none, the file backend is shut down instead of quietly filling a log directory. Set LOG_SINK_PI_FILE_FALLBACK=1 to keep it.

Failures in the extension itself warn to the console and never propagate: telemetry that breaks the agent it is measuring is worse than no telemetry.

Once a session has run, query it like any other service:

# What went wrong in this agent session
log-sink-mcp logs agent-issues --session-id <session-id>

# Token spend and per-tool consumption for the session
log-sink-mcp logs metrics --service pi --group-by agent

Integration Examples

Pino Logger

import pino from 'pino';

const logger = pino({
  transport: {
    target: 'pino-http-send',
    options: {
      url: 'http://localhost:3100/logs',
      method: 'POST',
    },
  },
});

logger.info({ traceId: 'abc123', userId: '456' }, 'User logged in');

Winston Logger

import winston from 'winston';

const logger = winston.createLogger({
  transports: [
    new winston.transports.Http({
      host: 'localhost',
      port: 3100,
      path: '/logs',
    }),
  ],
});

logger.error('Database connection failed', {
  service: 'api',
  errorType: 'ConnectionError'
});

Custom Fetch

async function sendLog(level: string, message: string, metadata?: any) {
  await fetch('http://localhost:3100/logs', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      logs: [{
        level,
        message,
        service: 'my-service',
        metadata,
        timestamp: new Date().toISOString(),
      }],
    }),
  });
}

await sendLog('info', 'Request processed', { duration: 245, status: 200 });

Configuration

CLI Options

start command (Recommended)

Start HTTP and/or MCP servers with automatic singleton coordination.

bun run src/cli.ts start [options]

Options:

• --port <number>: HTTP server port (default: 3100)
• --db-path <path>: SQLite database path (default: ./logs/session.db)
• --in-memory: Use in-memory database for testing (default: false)
• --http-only: Start only the HTTP server
• --mcp-only: Start only the MCP server (stdio transport)

Examples:

# Start both servers with defaults
bun run src/cli.ts start

# Production mode with persistent database
bun run src/cli.ts start --port 3100 --db-path ./logs/production.db

# Testing with in-memory database
bun run src/cli.ts start --in-memory

status command

Show diagnostic information about running services.

bun run src/cli.ts status

Displays:

• HTTP server status (running/stopped)
• Process ID and port
• Database file path and size
• Health check URL

stop command

Gracefully stop HTTP server and cleanup registry/PID files.

bun run src/cli.ts stop

Behavior:

• Sends SIGTERM to HTTP server process
• Removes .agiflow/registry/ entries
• Cleans up .pids/ files

http-serve command

Manually start HTTP log ingestion server (for advanced use).

bun run src/cli.ts http-serve [options]

Options:

• --port <number>: HTTP server port (default: 3000)
• --db-path <path>: SQLite database path (default: ./logs.db)
• --in-memory: Use in-memory database (default: false)

Note: Prefer using start command for automatic singleton coordination.

mcp-serve command

Manually start MCP server (for advanced use or custom transports).

bun run src/cli.ts mcp-serve [options]

Options:

• --type <type>: Transport type: stdio, http, or sse (default: stdio)
• --port <number>: Port for http/sse transport (default: 3000)
• --host <host>: Host for http/sse transport (default: localhost)
• --db-path <path>: SQLite database path (default: ./logs/session.db)
• --in-memory: Use in-memory database (default: false)
• --cleanup: Stop HTTP server on MCP shutdown (stdio only, default: false)

Examples:

# stdio transport (for Claude Desktop)
bun run src/cli.ts mcp-serve --type stdio

# HTTP transport
bun run src/cli.ts mcp-serve --type http --port 3001

# SSE transport
bun run src/cli.ts mcp-serve --type sse --port 3002

Note: For stdio transport, prefer using start --mcp-only for automatic HTTP coordination.

logs command

Run log analysis capabilities directly from the shell. Most subcommands mirror an MCP tool; analyze-errors and clear are CLI-only.

bun run src/cli.ts logs <subcommand> [options]

Subcommands:

• query: Compatibility alias for filter-only search_logs
• search: Unified filter/search command; the text query is optional
• search --mode hybrid: Default hybrid search across FTS5 and semantic embeddings
• trace: Equivalent to get_trace_timeline
• analyze-errors: CLI-only grouped error analysis
• agent-issues: Equivalent to analyze_agent_issues
• stats: Equivalent to get_log_stats
• metrics: Workflow, token, failure, and per-tool consumption metrics
• services: Equivalent to get_services
• clear: CLI-only command to delete all logs from the selected database

All subcommands support:

• --db-path <path>: SQLite database path (default: ./logs/session.db)
• --in-memory: Use an in-memory database

Examples:

# Query error logs
bun run src/cli.ts logs query --level error --db-path ./logs/session.db

# Search for timeout failures
bun run src/cli.ts logs search "timeout OR retry" --service api-gateway user-service

# Inspect one trace
bun run src/cli.ts logs trace a1b2c3d4e5f6789012345678901234ab

# Analyze grouped errors in a time window
bun run src/cli.ts logs analyze-errors \
  --start-time 2026-01-15T10:00:00Z \
  --end-time 2026-01-15T11:00:00Z

# Inspect workflow and per-tool token metrics, narrowed to one provider/model
bun run src/cli.ts logs metrics \
  --provider openai \
  --model gpt-5.3-codex \
  --sort total-tokens \
  --tool-sort p90-total-tokens

# Compare workflow metrics by model or provider
bun run src/cli.ts logs metrics --group-by model
bun run src/cli.ts logs metrics --group-by provider

# View service names discovered in the log database
bun run src/cli.ts logs services

Environment Variables

None are required: the server and its CLI are configured entirely through flags. These apply to the telemetry side, meaning the Pi extension and any process using @agimon-ai/log-sink-mcp/telemetry/node.

Variable	Effect
LOG_SINK_ENDPOINT	Send to this log-sink instead of discovering a running one
OTEL_EXPORTER_OTLP_ENDPOINT	Standard OTLP endpoint; the signal-specific _LOGS_ and _TRACES_ forms take precedence, and any of them override discovery
OTEL_EXPORTER_OTLP_HEADERS	Extra export headers; _LOGS_ and _TRACES_ variants apply per signal
OTEL_EXPORTER_OTLP_TIMEOUT	Export timeout in milliseconds; _LOGS_ and _TRACES_ variants apply per signal
OTEL_SDK_DISABLED	Disable telemetry entirely
AGENT_TELEMETRY_DISABLED=1	Disable the Pi extension only
AGENT_OTEL_TRACES	1, true, or yes to export spans as well as logs
AGENT_OTEL_REDACT=0	Include prompts, tool inputs, and tool results
LOG_SINK_PI_FILE_FALLBACK=1	Let the Pi extension write to files when no sink is reachable
LOG_SINK_FILE_MAX_BYTES, LOG_SINK_FILE_MAX_FILES	Rotation limits for the file backend

Telemetry is tagged for later filtering from the environment it runs in. AGENT_SESSION_ID and PARENT_AGENT_SESSION_ID scope records to an agent launch, and WORKFLOW_RUN_ID, WORKFLOW_RUN_KEY, WORKFLOW_ID, WORKFLOW_NAME, WORKFLOW_WORKSPACE, WORKFLOW_JOB_ID, WORKFLOW_JOB_NAME, WORKFLOW_STEP_NAME, WORKFLOW_PHASE, and WORKFLOW_RUNNER scope them to a workflow run. Each becomes an ingest header and then a queryable field, which is what makes --workflow-run-id and --session-id work across the CLI and MCP tools.

Development

Setup

# Install dependencies
pnpm install

# Run development server
pnpm dev

# Run HTTP server in dev mode
bun run src/cli.ts http-serve --port 3100

Project Structure

log-sink-mcp/
├── src/
│   ├── cli.ts                    # CLI entry point
│   ├── commands/                 # CLI commands
│   │   ├── http-serve.ts        # HTTP server command
│   │   ├── mcp-serve.ts         # MCP server command
│   │   ├── logs.ts              # Log analysis subcommands
│   │   └── claude-hook.ts       # Claude Code hook context command
│   ├── container/               # InversifyJS DI container
│   ├── extensions/              # Pi extension
│   │   ├── pi.ts                # Extension entry point
│   │   └── piTelemetry.ts       # Pi session event to OTLP mapping
│   ├── telemetry/
│   │   └── node.ts              # Node OTLP setup, endpoint discovery, file fallback
│   ├── models/                  # Drizzle schemas
│   │   └── schema.ts            # Log table schema
│   ├── server/
│   │   ├── http.ts              # Hono HTTP server
│   │   └── index.ts             # MCP server
│   ├── services/                # Business logic
│   │   ├── LogStorageService.ts
│   │   ├── LogQueryService.ts
│   │   ├── LogSearchService.ts
│   │   └── LogRetentionService.ts
│   ├── tools/                   # MCP and CLI tool implementations
│   │   ├── SearchLogsTool.ts
│   │   ├── QueryLogsTool.ts     # Compatibility wrapper for filter-only queries
│   │   ├── GetTraceTimelineTool.ts
│   │   ├── AnalyzeErrorsTool.ts
│   │   ├── AnalyzeAgentIssuesTool.ts
│   │   ├── GetLogStatsTool.ts
│   │   ├── GetServicesTool.ts
│   │   └── ClearLogsTool.ts
│   └── types/                   # TypeScript types
├── migrations/                  # Drizzle migrations
├── tests/
│   ├── integration/             # Integration tests
│   ├── services/                # Service unit tests
│   ├── tools/                   # Tool unit tests
│   └── commands/                # Command tests
└── README.md

Testing

# Run all tests
pnpm test

# Run integration tests only
pnpm test integration

# Run with coverage
pnpm test --coverage

Test Coverage:

• HTTP server integration tests (8 tests)
• MCP/tool integration tests
• Command tests (5 tests)

Adding New Tools

1. Create tool class in src/tools/
2. Implement Tool interface with getDefinition() and execute()
3. Register in src/server/index.ts
4. Add integration tests in tests/integration/

See existing tools for patterns.

Troubleshooting

Check Service Status

Use the status command to diagnose issues:

bun run src/cli.ts status

This shows:

• Whether HTTP server is running
• Process ID and port
• Database path and size
• Health check URL

Logs not appearing

1. Check HTTP server is running:

   # Using status command
   bun run src/cli.ts status
   
   # Or direct health check
   curl http://localhost:3100/health
   

2. Verify log format matches API schema

3. Check database path is accessible

4. Review HTTP server logs for errors

MCP tools not working

1. Verify database file exists and has logs:

   bun run src/cli.ts status
   

2. Check --db-path matches between HTTP and MCP servers

3. Restart MCP server after configuration changes:

   bun run src/cli.ts stop
   bun run src/cli.ts start
   

Database locked errors

SQLite uses WAL mode for better concurrency, but:

• Don't run multiple HTTP servers on same database
• Ensure file permissions allow writes
• Use separate databases for testing

Finding and Killing HTTP Server

If you need to manually stop the HTTP server:

# Graceful stop (recommended)
bun run src/cli.ts stop

# Find process manually
cat .pids/log-sink-mcp-http-development.pid

# Kill process (if graceful stop fails)
kill $(cat .pids/log-sink-mcp-http-development.pid)

# Force kill (last resort)
kill -9 $(cat .pids/log-sink-mcp-http-development.pid)

Stale Registry Entries

If the registry file exists but server isn't running:

# Clean up stale registry
rm .agiflow/registry/log-sink-mcp-http_development.json
rm .pids/log-sink-mcp-http-development.pid

# Restart
bun run src/cli.ts start

The health check will automatically detect stale registries and start a new server.

Port Already in Use

If port 3100 is already taken:

# Use different port
bun run src/cli.ts start --port 3101

# Or find what's using the port
lsof -i :3100

# Kill the process using the port
kill $(lsof -t -i :3100)

License

MIT

Contributing

This package is part of the Agiflow monorepo. Contributions welcome!

1. Follow coding standards in /docs/CODING_STANDARDS.md
2. Use architect MCP before modifying code
3. Add tests for new features
4. Run pnpm test before committing