This is page 1 of 2. Use http://codebase.md/brendancopley/mcp-chain-of-draft-prompt-tool?page={x} to view the full context.
# Directory Structure
```
├── __init__.py
├── .env.example
├── .gitignore
├── .npmrc
├── claude_desktop_config.example.json
├── commit.sh
├── config.json
├── Dockerfile
├── docs
│ ├── CLAUDE.md
│ └── mcp-documentation.md
├── nx.json
├── package-lock.json
├── package.json
├── project.json
├── promptsConfig.json
├── push.sh
├── README.md
├── RELEASE_NOTES.md
├── requirements.txt
├── sea-config.json
├── smithery.yaml
├── src
│ ├── index.ts
│ ├── python
│ │ ├── analytics.py
│ │ ├── client.py
│ │ ├── complexity.py
│ │ ├── examples.py
│ │ ├── format.py
│ │ ├── reasoning.py
│ │ └── server.py
│ ├── server.ts
│ ├── test-query.ts
│ ├── types.ts
│ └── utils
│ └── logger.ts
├── system_template.md
├── test.py
└── tsconfig.json
```
# Files
--------------------------------------------------------------------------------
/.npmrc:
--------------------------------------------------------------------------------
```
registry="https://registry.npmjs.org/"
@mcp-chain-of-draft-prompt-tool:registry="https://registry.npmjs.org/"
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Environment variables and configuration
.env
claude_desktop_config.json
# Python artifacts
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
*.egg-info/
.installed.cfg
*.egg
# Virtual environment
venv/
.venv/
ENV/
# Database files
*.db
*.sqlite
*.sqlite3
# Logs
*.log
# System files
.DS_Store
Thumbs.db
node_modules/*
dist/*
# Binary builds
binaries/
*.exe
# Nx specific
.nx/
.nxignore
.nx-cache/
# VSCode specific
.vscode/*
!.vscode/settings.json
!.vscode/tasks.json
!.vscode/launch.json
!.vscode/extensions.json
*.code-workspace
.history/
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
# LLM Provider Configuration
# Choose which provider to use: 'anthropic', 'openai', 'mistral', or 'ollama'
LLM_PROVIDER=anthropic
# Default model to use (provider-specific)
# Anthropic: claude-3-7-sonnet-latest, claude-3-opus-20240229
# OpenAI: gpt-4-turbo-preview, gpt-4, gpt-3.5-turbo
# Mistral: mistral-tiny, mistral-small, mistral-medium
# Ollama: llama2, mistral, codellama
LLM_MODEL=claude-3-7-sonnet-latest
# Anthropic Configuration
ANTHROPIC_API_KEY=your_anthropic_api_key_here
ANTHROPIC_BASE_URL=https://api.anthropic.com
# OpenAI Configuration
OPENAI_API_KEY=your_openai_api_key_here
OPENAI_BASE_URL=https://api.openai.com
# Mistral Configuration
MISTRAL_API_KEY=your_mistral_api_key_here
# Ollama Configuration (local deployment)
OLLAMA_BASE_URL=http://localhost:11434
# Chain of Draft Settings
# These are optional and will use defaults if not set
# MAX_WORDS_PER_STEP=5
# ENFORCE_FORMAT=true
# ADAPTIVE_WORD_LIMIT=true
# Database Settings
COD_DB_URL=sqlite:///cod_analytics.db
COD_EXAMPLES_DB=cod_examples.db
# Default Settings
COD_DEFAULT_MODEL=claude-3-7-sonnet-latest
COD_MAX_TOKENS=500
COD_MAX_WORDS_PER_STEP=5
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
[](https://mseep.ai/app/brendancopley-mcp-chain-of-draft-prompt-tool)
# MCP Chain of Draft (CoD) Prompt Tool
[](https://smithery.ai/server/@brendancopley/mcp-chain-of-draft-prompt-tool)
[](https://npmjs.org/mcp-chain-of-draft-prompt-tool)
[](https://packagephobia.com/result?p=mcp-chain-of-draft-prompt-tool)
[](https://github.com/brendancopley/mcp-chain-of-draft-prompt-tool/blob/master/LICENSE)
[](https://github.com/brendancopley/mcp-chain-of-draft-prompt-tool/stargazers)
[](https://github.com/brendancopley/mcp-chain-of-draft-prompt-tool/fork)
[](https://mseep.ai/app/c7233b8a-33e4-492d-8840-e3bddcf48aa3)
## Overview
The MCP Chain of Draft (CoD) Prompt Tool is a powerful Model Context Protocol tool that enhances LLM reasoning by transforming standard prompts into either Chain of Draft (CoD) or Chain of Thought (CoT) format. Here's how it works:
1. **Input Transformation**: Your regular prompt is automatically transformed into a CoD/CoT format
2. **LLM Processing**: The transformed prompt is passed to your chosen LLM (Claude, GPT, Ollama, or local models)
3. **Enhanced Reasoning**: The LLM processes the request using structured reasoning steps
4. **Result Transformation**: The response is transformed back into a clear, concise format
This approach significantly improves reasoning quality while reducing token usage and maintaining high accuracy.
## BYOLLM Support
This tool supports a "Bring Your Own LLM" approach, allowing you to use any language model of your choice:
### Supported LLM Integrations
- **Cloud Services**
- Anthropic Claude
- OpenAI GPT models
- Mistral AI
- **Local Models**
- Ollama (all models)
- Local LLama variants
- Any model supporting chat completion API
### Configuring Your LLM
1. **Cloud Services**
```bash
# For Anthropic Claude
export ANTHROPIC_API_KEY=your_key_here
# For OpenAI
export OPENAI_API_KEY=your_key_here
# For Mistral AI
export MISTRAL_API_KEY=your_key_here
```
2. **Local Models with Ollama**
```bash
# First install Ollama
curl https://ollama.ai/install.sh | sh
# Pull your preferred model
ollama pull llama2
# or
ollama pull mistral
# or any other model
# Configure the tool to use Ollama
export MCP_LLM_PROVIDER=ollama
export MCP_OLLAMA_MODEL=llama2 # or your chosen model
```
3. **Custom Local Models**
```bash
# Point to your local model API
export MCP_LLM_PROVIDER=custom
export MCP_CUSTOM_LLM_ENDPOINT=http://localhost:your_port
```
## Credits
This project implements the Chain of Draft (CoD) reasoning approach as a Model Context Protocol (MCP) prompt tool for Claude. The core Chain of Draft implementation is based on the work by [stat-guy](https://github.com/stat-guy/chain-of-draft). We extend our gratitude for their pioneering work in developing this efficient reasoning approach.
Original Repository: [https://github.com/stat-guy/chain-of-draft](https://github.com/stat-guy/chain-of-draft)
## Key Benefits
- **Efficiency**: Significantly reduced token usage (as little as 7.6% of standard CoT)
- **Speed**: Faster responses due to shorter generation time
- **Cost Savings**: Lower API costs for LLM calls
- **Maintained Accuracy**: Similar or even improved accuracy compared to CoT
- **Flexibility**: Applicable across various reasoning tasks and domains
## Features
1. **Core Chain of Draft Implementation**
- Concise reasoning steps (typically 5 words or less)
- Format enforcement
- Answer extraction
2. **Performance Analytics**
- Token usage tracking
- Solution accuracy monitoring
- Execution time measurement
- Domain-specific performance metrics
3. **Adaptive Word Limits**
- Automatic complexity estimation
- Dynamic adjustment of word limits
- Domain-specific calibration
4. **Comprehensive Example Database**
- CoT to CoD transformation
- Domain-specific examples (math, code, biology, physics, chemistry, puzzle)
- Example retrieval based on problem similarity
5. **Format Enforcement**
- Post-processing to ensure adherence to word limits
- Step structure preservation
- Adherence analytics
6. **Hybrid Reasoning Approaches**
- Automatic selection between CoD and CoT
- Domain-specific optimization
- Historical performance-based selection
7. **OpenAI API Compatibility**
- Drop-in replacement for standard OpenAI clients
- Support for both completions and chat interfaces
- Easy integration into existing workflows
## Setup and Installation
### Prerequisites
- Python 3.10+ (for Python implementation)
- Node.js 22+ (for JavaScript implementation)
- Nx (for building Single Executable Applications)
### Python Installation
1. Clone the repository
2. Install dependencies:
```bash
pip install -r requirements.txt
```
3. Configure API keys in `.env` file:
```
ANTHROPIC_API_KEY=your_api_key_here
```
4. Run the server:
```bash
python server.py
```
### JavaScript/TypeScript Installation
1. Clone the repository
2. Install dependencies:
```bash
npm install
```
3. Configure API keys in `.env` file:
```
ANTHROPIC_API_KEY=your_api_key_here
```
4. Build and run the server:
```bash
# Build TypeScript files using Nx
npm run nx build
# Start the server
npm start
# For development with auto-reload:
npm run dev
```
Available scripts:
- `npm run nx build`: Compiles TypeScript to JavaScript using Nx build system
- `npm run build:sea`: Creates Single Executable Applications for all platforms
- `npm start`: Runs the compiled server from `dist`
- `npm test`: Runs the test query against the server
- `npm run dev`: Runs the TypeScript server directly using ts-node (useful for development)
The project uses Nx as its build system, providing:
- Efficient caching and incremental builds
- Cross-platform build support
- Integrated SEA generation
- Dependency graph visualization
- Consistent build process across environments
## Single Executable Applications (SEA)
This project supports building Single Executable Applications (SEA) using Node.js 22+ and the [@getlarge/nx-node-sea](https://github.com/getlarge/nx-node-sea) plugin. This allows you to create standalone executables that don't require Node.js to be installed on the target system.
### Building SEA Executables
The project includes several scripts for building SEA executables:
```bash
# Build for all platforms
npm run build:sea
# Build for specific platforms
npm run build:macos # macOS
npm run build:linux # Linux
npm run build:windows # Windows
```
### SEA Build Configuration
The project uses Nx for managing the build process. The SEA configuration is handled through the nx-node-sea plugin, which provides a streamlined way to create Node.js single executable applications.
Key features of the SEA build process:
- Cross-platform support (macOS, Linux, Windows)
- Automatic dependency bundling
- Optimized binary size
- No runtime dependencies required
### Using SEA Executables
Once built, the SEA executables can be found in the `dist` directory. These executables:
- Are completely standalone
- Don't require Node.js installation
- Can be distributed and run directly
- Maintain all functionality of the original application
For Claude Desktop integration with SEA executables, update your configuration to use the executable path:
```json
{
"mcpServers": {
"chain-of-draft-prompt-tool": {
"command": "/path/to/mcp-chain-of-draft-prompt-tool",
"env": {
"ANTHROPIC_API_KEY": "your_api_key_here"
}
}
}
}
```
## Claude Desktop Integration
To integrate with Claude Desktop:
1. Install Claude Desktop from [claude.ai/download](https://claude.ai/download)
2. Create or edit the Claude Desktop config file:
```
~/Library/Application Support/Claude/claude_desktop_config.json
```
3. Add the tool configuration (Python version):
```json
{
"mcpServers": {
"chain-of-draft-prompt-tool": {
"command": "python3",
"args": ["/absolute/path/to/cod/server.py"],
"env": {
"ANTHROPIC_API_KEY": "your_api_key_here"
}
}
}
}
```
Or for the JavaScript version:
```json
{
"mcpServers": {
"chain-of-draft-prompt-tool": {
"command": "node",
"args": ["/absolute/path/to/cod/index.js"],
"env": {
"ANTHROPIC_API_KEY": "your_api_key_here"
}
}
}
}
```
4. Restart Claude Desktop
You can also use the Claude CLI to add the tool:
```bash
# For Python implementation
claude mcp add chain-of-draft-prompt-tool -e ANTHROPIC_API_KEY="your_api_key_here" "python3 /absolute/path/to/cod/server.py"
# For JavaScript implementation
claude mcp add chain-of-draft-prompt-tool -e ANTHROPIC_API_KEY="your_api_key_here" "node /absolute/path/to/cod/index.js"
```
## Using with Dive GUI
[Dive](https://github.com/OpenAgentPlatform/Dive) is an excellent open-source MCP Host Desktop Application that provides a user-friendly GUI for interacting with MCP tools like this one. It supports multiple LLMs including ChatGPT, Anthropic Claude, Ollama, and other OpenAI-compatible models.
### Integrating with Dive
1. Download and install Dive from their [releases page](https://github.com/OpenAgentPlatform/Dive/releases)
2. Configure the Chain of Draft tool in Dive's MCP settings:
```json
{
"mcpServers": {
"chain-of-draft-prompt-tool": {
"command": "/path/to/mcp-chain-of-draft-prompt-tool",
"enabled": true,
"env": {
"ANTHROPIC_API_KEY": "your_api_key_here"
}
}
}
}
```
If you're using the non-SEA version:
```json
{
"mcpServers": {
"chain-of-draft-prompt-tool": {
"command": "node",
"args": ["/path/to/dist/index.js"],
"enabled": true,
"env": {
"ANTHROPIC_API_KEY": "your_api_key_here"
}
}
}
}
```
### Key Benefits of Using Dive
- 🌐 Universal LLM Support with multiple API key management
- 💻 Cross-platform availability (Windows, MacOS, Linux)
- 🔄 Seamless MCP integration in both stdio and SSE modes
- 🌍 Multi-language interface
- 💡 Custom instructions and system prompts
- 🔄 Automatic updates
Using Dive provides a convenient way to interact with the Chain of Draft tool through a modern, feature-rich interface while maintaining all the benefits of the MCP protocol.
## Testing with MCP Inspector
The project includes integration with the MCP Inspector tool, which provides a visual interface for testing and debugging MCP tools. This is especially useful during development or when you want to inspect the tool's behavior.
### Running the Inspector
You can start the MCP Inspector using the provided npm script:
```bash
# Start the MCP Inspector with the tool
npm run test-inspector
# Or run it manually
npx @modelcontextprotocol/inspector -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY -- node dist/index.js
```
This will:
1. Start the MCP server in the background
2. Launch the MCP Inspector interface in your default browser
3. Connect to the running server for testing
### Using the Inspector Interface
The MCP Inspector provides:
- 🔍 Real-time visualization of tool calls and responses
- 📝 Interactive testing of MCP functions
- 🔄 Request/response history
- 🐛 Debug information for each interaction
- 📊 Performance metrics and timing data
This makes it an invaluable tool for:
- Development and debugging
- Understanding tool behavior
- Testing different inputs and scenarios
- Verifying MCP compliance
- Performance optimization
The Inspector will be available at `http://localhost:5173` by default.
## Available Tools
The Chain of Draft server provides the following tools:
| Tool | Description |
|------|-------------|
| `chain_of_draft_solve` | Solve a problem using Chain of Draft reasoning |
| `math_solve` | Solve a math problem with CoD |
| `code_solve` | Solve a coding problem with CoD |
| `logic_solve` | Solve a logic problem with CoD |
| `get_performance_stats` | Get performance stats for CoD vs CoT |
| `get_token_reduction` | Get token reduction statistics |
| `analyze_problem_complexity` | Analyze problem complexity |
## Developer Usage
### Python Client
If you want to use the Chain of Draft client directly in your Python code:
```python
from client import ChainOfDraftClient
# Create client with specific LLM provider
cod_client = ChainOfDraftClient(
llm_provider="ollama", # or "anthropic", "openai", "mistral", "custom"
model_name="llama2" # specify your model
)
# Use directly
result = await cod_client.solve_with_reasoning(
problem="Solve: 247 + 394 = ?",
domain="math"
)
print(f"Answer: {result['final_answer']}")
print(f"Reasoning: {result['reasoning_steps']}")
print(f"Tokens used: {result['token_count']}")
```
### JavaScript/TypeScript Client
For TypeScript/Node.js applications:
```typescript
import { ChainOfDraftClient } from './lib/chain-of-draft-client';
// Create client with your preferred LLM
const client = new ChainOfDraftClient({
provider: 'ollama', // or 'anthropic', 'openai', 'mistral', 'custom'
model: 'llama2', // your chosen model
endpoint: 'http://localhost:11434' // for custom endpoints
});
// Use the client
async function solveMathProblem() {
const result = await client.solveWithReasoning({
problem: "Solve: 247 + 394 = ?",
domain: "math",
max_words_per_step: 5
});
console.log(`Answer: ${result.final_answer}`);
console.log(`Reasoning: ${result.reasoning_steps}`);
console.log(`Tokens used: ${result.token_count}`);
}
solveMathProblem();
```
## Implementation Details
The server is available in both Python and JavaScript implementations, both consisting of several integrated components:
### Python Implementation
1. **AnalyticsService**: Tracks performance metrics across different problem domains and reasoning approaches
2. **ComplexityEstimator**: Analyzes problems to determine appropriate word limits
3. **ExampleDatabase**: Manages and retrieves examples, transforming CoT examples to CoD format
4. **FormatEnforcer**: Ensures reasoning steps adhere to word limits
5. **ReasoningSelector**: Intelligently chooses between CoD and CoT based on problem characteristics
### JavaScript Implementation
1. **analyticsDb**: In-memory database for tracking performance metrics
2. **complexityEstimator**: Analyzes problems to determine complexity and appropriate word limits
3. **formatEnforcer**: Ensures reasoning steps adhere to word limits
4. **reasoningSelector**: Automatically chooses between CoD and CoT based on problem characteristics and historical performance
Both implementations follow the same core principles and provide identical MCP tools, making them interchangeable for most use cases.
## License
This project is open-source and available under the MIT license.
```
--------------------------------------------------------------------------------
/docs/CLAUDE.md:
--------------------------------------------------------------------------------
```markdown
# Chain of Draft MCP Server
## Overview
- This repository implements a Chain of Draft (CoD) reasoning server
- CoD is a method for efficient reasoning with LLMs using minimal word steps
## Dependencies
- Required packages are in `requirements.txt`
- Install with `pip install -r requirements.txt`
- Requires an Anthropic API key in `.env` file
## Running the Server
- Start the server with `python server.py`
- Alternatively, configure in Claude Desktop config
## Testing
- Run tests with `python test.py`
## Claude Desktop Integration
- Add to `~/Library/Application Support/Claude/claude_desktop_config.json`
- Point to the absolute path of server.py
- Include your Anthropic API key in the environment
- Restart Claude Desktop after configuration
## Available Tools
- `chain_of_draft_solve`: General problem solver with CoD
- `math_solve`: Specifically for math problems
- `code_solve`: For coding problems
- `logic_solve`: For logic problems
- `get_performance_stats`: View statistics on CoD vs CoT
- `get_token_reduction`: Analyze token usage reduction
- `analyze_problem_complexity`: Evaluate problem difficulty
## Pushing to GitHub
- Use `./push.sh` to push changes to GitHub
- Use `./commit.sh` to commit changes
```
--------------------------------------------------------------------------------
/push.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
git push -u origin main
```
--------------------------------------------------------------------------------
/commit.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
git commit -m "Initial commit: Chain of Draft MCP server implementation"
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
mcp>=1.2.0
anthropic>=0.15.0
openai>=1.12.0
mistralai>=0.0.12
sqlalchemy>=2.0.0
python-dotenv>=1.0.0
uuid>=1.30
aiohttp>=3.8.0
uvicorn>=0.23.0
pydantic>=2.0.0
fastapi>=0.100.0
typing-extensions>=4.5.0
asyncio>=3.4.3
```
--------------------------------------------------------------------------------
/claude_desktop_config.example.json:
--------------------------------------------------------------------------------
```json
{
"mcpServers": {
"mcp-chain-of-draft-prompt-tool": {
"command": "python",
"args": ["-m", "server.server"],
"env": {
"ANTHROPIC_API_KEY": "your_anthropic_api_key_here"
}
}
}
}
```
--------------------------------------------------------------------------------
/sea-config.json:
--------------------------------------------------------------------------------
```json
{
"main": "dist/index.js",
"output": "binaries/mcp-chain-of-draft-prompt-tool",
"assets": [
"dist/**/*",
"config.json",
"promptsConfig.json",
"system_template.md"
],
"env": {
"NODE_ENV": "production"
},
"platforms": ["darwin", "linux", "win32"],
"architectures": {
"darwin": ["x64", "arm64"],
"linux": ["x64", "arm64"],
"win32": ["x64"]
}
}
```
--------------------------------------------------------------------------------
/tsconfig.json:
--------------------------------------------------------------------------------
```json
{
"compilerOptions": {
"target": "ES2020",
"module": "ES2020",
"moduleResolution": "node",
"esModuleInterop": true,
"strict": true,
"outDir": "dist/",
"rootDir": "src/",
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true
},
"include": ["src/**/*.ts", "src/index.ts", "src/server.tssrc/test-query.tsquery.ts", "src/types.ts"],
"exclude": ["node_modules", "dist"]
}
```
--------------------------------------------------------------------------------
/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Chain of Draft (CoD) MCP Server.
This package implements the Chain of Draft reasoning approach
as described in the paper "Chain of Draft: Thinking Faster by Writing Less".
"""
from .analytics import AnalyticsService
from .complexity import ComplexityEstimator
from .examples import ExampleDatabase
from .format import FormatEnforcer
from .reasoning import ReasoningSelector, create_cod_prompt, create_cot_prompt
from .client import ChainOfDraftClient
__version__ = "0.1.0"
```
--------------------------------------------------------------------------------
/project.json:
--------------------------------------------------------------------------------
```json
{
"name": "mcp-chain-of-draft-prompt-tool",
"$schema": "node_modules/nx/schemas/project-schema.json",
"sourceRoot": "src",
"projectType": "application",
"plugins": [
{
"plugin": "@getlarge/nx-node-sea",
"options": {
"seaTargetName": "sea-build",
"buildTarget": "build"
}
}
],
"targets": {
"build": {
"executor": "@nx/js:tsc",
"outputs": ["{options.outputPath}"],
"options": {
"outputPath": "dist",
"main": "src/index.ts",
"tsConfig": "tsconfig.json",
"assets": ["package.json"]
}
}
}
}
```
--------------------------------------------------------------------------------
/nx.json:
--------------------------------------------------------------------------------
```json
{
"extends": "nx/presets/npm.json",
"$schema": "./node_modules/nx/schemas/nx-schema.json",
"namedInputs": {
"default": ["{projectRoot}/**/*"],
"production": ["default"]
},
"plugins": [
{
"plugin": "@getlarge/nx-node-sea",
"options": {
"seaTargetName": "sea-build",
"buildTarget": "build"
}
}
],
"tasksRunnerOptions": {
"default": {
"runner": "nx/tasks-runners/default",
"options": {
"cacheableOperations": ["build", "test"]
}
}
},
"targetDefaults": {
"build": {
"dependsOn": ["^build"],
"inputs": [],
"options": {
"outputPath": "dist",
"main": "src/index.ts",
"tsConfig": "tsconfig.json",
"rootDir": "src",
"outputFileName": "index.js"
}
}
}
}
```
--------------------------------------------------------------------------------
/config.json:
--------------------------------------------------------------------------------
```json
{
"server": {
"name": "MCP Chain of Draft (CoD) Prompt Tool",
"version": "1.0.0",
"port": 3000,
"description": "A server for Chain of Draft reasoning and analysis"
},
"prompts": {
"file": "promptsConfig.json",
"registrationMode": "name",
"categories": ["chain-of-draft", "analytics"]
},
"transports": {
"default": "stdio",
"sse": {
"enabled": false
},
"stdio": {
"enabled": true,
"debug": true
}
},
"logging": {
"directory": "./logs",
"level": "info",
"format": "json",
"transport": true
},
"analytics": {
"databases": {
"examples": "cod_examples.db",
"analytics": "cod_analytics.db"
}
},
"features": {
"adaptiveWordLimit": true,
"formatEnforcement": true,
"complexityAnalysis": true,
"performanceTracking": true
}
}
```
--------------------------------------------------------------------------------
/src/utils/logger.ts:
--------------------------------------------------------------------------------
```typescript
import chalk from 'chalk';
export const logger = {
info: (message: string, ...args: any[]) => {
console.log(chalk.blue('ℹ'), chalk.blue(message), ...args);
},
success: (message: string, ...args: any[]) => {
console.log(chalk.green('✓'), chalk.green(message), ...args);
},
warning: (message: string, ...args: any[]) => {
console.warn(chalk.yellow('⚠'), chalk.yellow(message), ...args);
},
error: (message: string, ...args: any[]) => {
console.error(chalk.red('✖'), chalk.red(message), ...args);
},
debug: (message: string, ...args: any[]) => {
console.log(chalk.gray('🔍'), chalk.gray(message), ...args);
},
highlight: (message: string, ...args: any[]) => {
console.log(chalk.cyan('→'), chalk.cyan(message), ...args);
},
devLog: (message: string, ...args: any[]) => {
if (process.env.NODE_ENV === 'development') {
console.log(chalk.gray('[DEV]'), chalk.gray(message), ...args);
}
},
// Special formatting for Chain of Draft outputs
codOutput: {
header: (text: string) => console.log(chalk.bold.magenta('\n=== ' + text + ' ===\n')),
problem: (text: string) => console.log(chalk.yellow('Problem: ') + text),
steps: (steps: string) => console.log(chalk.cyan('Reasoning Steps:\n') + steps),
answer: (text: string) => console.log(chalk.green('\nFinal Answer: ') + text),
stats: (stats: Record<string, any>) => {
console.log(chalk.blue('\nStats:'));
Object.entries(stats).forEach(([key, value]) => {
console.log(chalk.blue(`- ${key}: `) + chalk.white(value));
});
}
}
};
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Build stage
FROM node:22.12-alpine AS builder
# Install only necessary build dependencies
RUN apk add --no-cache \
python3 \
py3-pip \
py3-virtualenv
WORKDIR /app
# Copy dependency files
COPY package*.json requirements.txt ./
# Setup .env file (use .env if it exists, otherwise use .env.example)
COPY .env* ./
RUN if [ ! -f .env ] && [ -f .env.example ]; then cp .env.example .env; fi
# Install Node.js dependencies with caching
RUN --mount=type=cache,target=/root/.npm \
npm ci --ignore-scripts --omit-dev
# Setup Python environment and install dependencies
RUN python3 -m venv /app/venv
ENV PATH="/app/venv/bin:$PATH"
RUN --mount=type=cache,target=/root/.cache/pip \
pip3 install --no-cache-dir -r requirements.txt
# Set build-time environment variables
ENV NODE_ENV=production
# Copy source and build
COPY . .
RUN npm run build
# Release stage
FROM node:22-alpine AS release
# Install runtime dependencies
RUN apk add --no-cache \
python3 \
py3-pip
WORKDIR /app
# Copy built artifacts and dependencies
COPY --from=builder /app/dist /app/dist
COPY --from=builder /app/package*.json /app/
COPY --from=builder /app/requirements.txt /app/
COPY --from=builder /app/venv /app/venv
COPY --from=builder /app/.env /app/.env
# Set runtime environment variables
ENV PATH="/app/venv/bin:$PATH" \
NODE_ENV=production
# Install production dependencies
RUN npm ci --omit=dev
# Set executable permissions
RUN chmod +x /app/dist/index.js
# Set production environment
ENV NODE_ENV=production
# Set the user to non-root for security
USER node
# Create startup script
ENTRYPOINT ["node", "/app/dist/index.js"]
```
--------------------------------------------------------------------------------
/src/test-query.ts:
--------------------------------------------------------------------------------
```typescript
import fetch from 'node-fetch';
import { logger } from './utils/logger.js';
interface ChainOfDraftResponse {
reasoning_steps: string;
final_answer: string;
approach: string;
stats: {
word_limit: number;
token_count: number;
execution_time_ms: number;
complexity: number;
};
}
function isChainOfDraftResponse(data: unknown): data is ChainOfDraftResponse {
return (
typeof data === 'object' &&
data !== null &&
'reasoning_steps' in data &&
'final_answer' in data &&
'approach' in data &&
'stats' in data
);
}
async function testChainOfDraft() {
try {
const problem = "What is the sum of the first 10 prime numbers?";
const response = await fetch('http://localhost:3000/solve', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
problem,
domain: 'math',
approach: 'CoD'
})
});
const rawData = await response.json();
if (!isChainOfDraftResponse(rawData)) {
throw new Error('Invalid response format from server');
}
const data: ChainOfDraftResponse = rawData;
logger.codOutput.header('Chain of Draft Demo');
logger.codOutput.problem(problem);
logger.codOutput.steps(data.reasoning_steps);
logger.codOutput.answer(data.final_answer);
logger.codOutput.stats({
'Approach': data.approach,
'Word limit': data.stats.word_limit,
'Tokens used': data.stats.token_count,
'Execution time': `${data.stats.execution_time_ms}ms`,
'Complexity score': data.stats.complexity
});
} catch (error) {
logger.error('Error testing Chain of Draft:', error);
}
}
testChainOfDraft();
```
--------------------------------------------------------------------------------
/system_template.md:
--------------------------------------------------------------------------------
```markdown
TEMPLATE """{{- if .Messages }}
{{- if or .System .Tools }}<start_of_turn>user
{{- if .System}}
{{ .System }}
{{- end }}
{{- if .Tools }}
# Function Calling Instructions
You are provided with a set of functions to help complete tasks. Here are your instructions for using them:
1. FUNCTION DEFINITIONS
The available functions are defined within <tools></tools> XML tags below:
<tools>
{{- range $.Tools }}
{"type": "function", "function": {{ .Function }}}
{{- end }}
</tools>
2. FUNCTION CALLING FORMAT
When calling a function:
- Use ```tool_use``` code blocks
- Include both function name and arguments as a JSON object
- Follow this exact format:
```tool_use
{"name": "<function-name>", "arguments": <args-json-object>}
```
3. GUIDELINES
- Analyze the task carefully before choosing functions
- Only call functions that are necessary
- Validate all required arguments before calling
- Handle function responses appropriately
- Maintain context between function calls
- Chain multiple functions if needed to complete the task
4. ERROR HANDLING
- Validate argument types and formats
- Check for missing required arguments
- Handle function response errors gracefully
- Retry with different approaches if needed
5. RESPONSE FORMAT
- Always provide clear explanations of your actions
- Format responses in a clear, structured way
- Include relevant context from function responses
- End with clear next steps or conclusions
{{- end }}<end_of_turn>
{{ end }}
{{- range $i, $_ := .Messages }}
{{- $last := eq (len (slice $.Messages $i)) 1 -}}
{{- if eq .Role "user" }}<start_of_turn>user
{{ .Content }}<end_of_turn>
{{ else if eq .Role "assistant" }}<start_of_turn>model
{{ if .Content }}{{ .Content }}
{{- else if .ToolCalls }}```tool_use
{{ range .ToolCalls }}{"name": "{{ .Function.Name }}", "arguments": {{ .Function.Arguments}}}
{{ end }}```
{{- end }}{{ if not $last }}<end_of_turn>
{{ end }}
{{- else if eq .Role "tool" }}<start_of_turn>user
<tool_response>
{{ .Content }}
</tool_response><end_of_turn>
{{ end }}
{{- if and (ne .Role "assistant") $last }}<start_of_turn>model
{{ end }}
{{- end }}
{{- else }}
{{- if .System }}<start_of_turn>user
{{ .System }}<end_of_turn>
{{ end }}{{ if .Prompt }}<start_of_turn>user
{{ .Prompt }}<end_of_turn>
{{ end }}<start_of_turn>model
{{ end }}{{ .Response }}{{ if .Response }}<end_of_turn>{{ end }}"""
```
--------------------------------------------------------------------------------
/package.json:
--------------------------------------------------------------------------------
```json
{
"name": "mcp-chain-of-draft-prompt-tool",
"version": "1.1.1",
"description": "MCP Chain of Draft (CoD) Prompt Tool - A Model Context Protocol tool for efficient reasoning",
"type": "module",
"engines": {
"node": ">=22"
},
"bin": {
"mcp-chain-of-draft-prompt-tool": "dist/index.js"
},
"files": [
"dist",
"README.md",
"LICENSE"
],
"scripts": {
"build": "tsc && shx chmod +x dist/*.js",
"build:docker": "docker build -t mcp-chain-of-draft-prompt-tool .",
"build:sea": "nx build && nx run-many --target=sea --all",
"build:macos": "nx build && nx run-many --target=sea --all --args=\"--platform=darwin\"",
"build:linux": "nx build && nx run-many --target=sea --all --args=\"--platform=linux\"",
"build:windows": "nx build && nx run-many --target=sea --all --args=\"--platform=win32\"",
"watch": "tsc --watch",
"start": "node dist/index.js",
"dev": "ts-node-esm dist/index.ts",
"test": "node dist/test-query.js",
"inspector": "npx @modelcontextprotocol/inspector -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY -- node dist/index.js",
"open-browser": "sleep 2 && open http://localhost:5173 || xdg-open http://localhost:5173 || start http://localhost:5173",
"test-inspector": "npm run start & sleep 2 && trap 'kill $!' SIGINT SIGTERM; (npm run inspector & npm run open-browser) || kill $!"
},
"keywords": [
"mcp",
"chain-of-draft",
"claude",
"reasoning",
"prompt-tool"
],
"author": "brendancopley",
"license": "MIT",
"repository": {
"type": "git",
"url": "git+https://github.com/brendancopley/mcp-chain-of-draft-prompt-tool.git"
},
"bugs": {
"url": "https://github.com/brendancopley/mcp-chain-of-draft-prompt-tool/issues"
},
"homepage": "https://github.com/brendancopley/mcp-chain-of-draft-prompt-tool#readme",
"dependencies": {
"@anthropic-ai/sdk": "^0.39.0",
"@mistralai/mistralai": "^1.5.2",
"@modelcontextprotocol/sdk": "1.7.0",
"chalk": "5",
"dotenv": "^16.4.5",
"node-fetch": "^3.3.2",
"ollama": "^0.5.14",
"openai": "^4.28.0"
},
"devDependencies": {
"@getlarge/nx-node-sea": "^0.2.0",
"@nx/devkit": "^20.6.2",
"@nx/js": "^20.6.2",
"@nx/workspace": "^20.6.2",
"@types/node": "^22",
"@types/node-fetch": "^2.6.11",
"nx": "^20.6.2",
"shx": "^0.4.0",
"ts-node": "^10.9.2",
"typescript": "^5.3.3"
},
"packageManager": "[email protected]+sha1.ac34549e6aa8e7ead463a7390f61a6610"
}
```
--------------------------------------------------------------------------------
/RELEASE_NOTES.md:
--------------------------------------------------------------------------------
```markdown
# Release Notes - v1.1.1
## MCP Chain of Draft Prompt Tool
We're excited to announce version 1.1.1 of the MCP Chain of Draft Prompt Tool! This release introduces BYOLLM support and improves documentation clarity.
### What's New
- **BYOLLM Support**: Added support for multiple LLM providers:
- Cloud services (Claude, GPT, Mistral)
- Local models via Ollama
- Custom local LLM endpoints
- **Pre-built Binaries**: Added standalone executables for all major platforms
- Enhanced documentation to better explain the tool's core functionality
- Improved README with clearer explanation of the transformation process
- Updated Smithery.ai integration and badges
### Core Functionality Highlight
The MCP Chain of Draft Prompt Tool serves as an intelligent prompt transformer that:
1. Takes your standard prompts
2. Automatically transforms them into Chain of Draft (CoD) or Chain of Thought (CoT) format
3. Processes them through your chosen LLM (cloud or local)
4. Returns enhanced, structured responses
This process significantly improves reasoning quality while reducing token usage.
### LLM Integration
The tool now supports multiple LLM providers:
#### Cloud Services
- Anthropic Claude
- OpenAI GPT models
- Mistral AI
#### Local Models
- Ollama integration (supporting all Ollama models)
- Custom local LLM endpoints
- Any model supporting chat completion API
### Pre-built Binaries
We now provide standalone executables for all major platforms:
#### macOS
- `mcp-chain-of-draft-prompt-tool-macos-arm64` (Apple Silicon)
- `mcp-chain-of-draft-prompt-tool-macos-x64` (Intel)
#### Linux
- `mcp-chain-of-draft-prompt-tool-linux-arm64` (ARM64)
- `mcp-chain-of-draft-prompt-tool-linux-x64` (x64)
#### Windows
- `mcp-chain-of-draft-prompt-tool-win-x64.exe`
### Integration
The tool is now available on Smithery.ai:
https://smithery.ai/server/@brendancopley/mcp-chain-of-draft-prompt-tool
### Getting Started
To get started with the new version:
```bash
# Using npm package
npm install [email protected]
# Or download the appropriate binary for your platform from the releases page
```
Configure your preferred LLM:
```bash
# For Ollama
export MCP_LLM_PROVIDER=ollama
export MCP_OLLAMA_MODEL=llama2
# For cloud services
export ANTHROPIC_API_KEY=your_key_here
# or
export OPENAI_API_KEY=your_key_here
# or
export MISTRAL_API_KEY=your_key_here
```
### Feedback
We welcome your feedback and contributions! Please visit our GitHub repository or Smithery.ai page to share your thoughts and experiences.
```
--------------------------------------------------------------------------------
/src/types.ts:
--------------------------------------------------------------------------------
```typescript
import { Anthropic } from '@anthropic-ai/sdk';
export interface AnalyticsRecord {
problem_id: string;
problem_text: string;
domain: string;
approach: string;
word_limit: number;
tokens_used: number;
execution_time_ms: number;
reasoning_steps: string;
answer: string;
timestamp?: Date;
}
export interface PerformanceStats {
domain: string;
approach: string;
avg_tokens: number;
avg_time_ms: number;
accuracy: number | null;
count: number;
}
export interface TokenReductionStats {
domain: string;
cod_avg_tokens: number;
cot_avg_tokens: number;
reduction_percentage: number;
}
export interface ComplexityAnalysis {
word_count: number;
sentence_count: number;
words_per_sentence: number;
indicator_count: number;
found_indicators: string[];
question_count: number;
estimated_complexity: number;
}
export interface DomainPreferences {
complexity_threshold: number;
accuracy_threshold: number;
}
export interface ChainOfDraftParams {
problem: string;
domain?: string;
max_words_per_step?: number | null;
approach?: string;
enforce_format?: boolean;
adaptive_word_limit?: boolean;
}
export interface ChainOfDraftResult {
approach: string;
reasoning_steps: string;
final_answer: string;
token_count: number;
word_limit: number;
complexity: number;
execution_time_ms: number;
}
export interface AnalyticsDatabase {
records: AnalyticsRecord[];
addRecord(record: AnalyticsRecord): number;
getPerformanceByDomain(domain?: string): PerformanceStats[];
getTokenReductionStats(): TokenReductionStats[];
}
export interface ComplexityEstimator {
domainBaseLimits: { [key: string]: number };
complexityIndicators: { [key: string]: string[] };
analyzeProblem(problem: string, domain: string): ComplexityAnalysis;
}
export interface FormatEnforcer {
enforceWordLimit(reasoning: string, maxWordsPerStep: number | null): string;
analyzeAdherence(reasoning: string, maxWordsPerStep: number): {
step_count: number;
total_words: number;
avg_words_per_step: number;
over_limit_steps: number;
adherence_percentage: number;
};
}
export interface ReasoningSelector {
defaultPreferences: { [key: string]: DomainPreferences };
selectApproach(domain: string, complexity: number, performanceStats: PerformanceStats[]): string;
}
export interface ChainOfDraftClient {
solveWithReasoning(params: ChainOfDraftParams): Promise<ChainOfDraftResult>;
}
export interface ToolArguments {
problem?: string;
domain?: string;
max_words_per_step?: number;
approach?: string;
enforce_format?: boolean;
adaptive_word_limit?: boolean;
}
```
--------------------------------------------------------------------------------
/test.py:
--------------------------------------------------------------------------------
```python
"""
Test script for the Chain of Draft client.
"""
import asyncio
import os
from dotenv import load_dotenv
# Use absolute import since this file might be run directly
from client import ChainOfDraftClient
# Load environment variables
load_dotenv()
async def main():
"""Run a simple test of the Chain of Draft client."""
# Create client
client = ChainOfDraftClient()
# Test problems
test_problems = [
{
"problem": "Jason had 20 lollipops. He gave Denny some lollipops. Now Jason has 12 lollipops. How many lollipops did Jason give to Denny?",
"domain": "math"
},
{
"problem": "A coin is heads up. John flips the coin. Mary flips the coin. Paul flips the coin. Susan does not flip the coin. Is the coin still heads up?",
"domain": "logic"
},
{
"problem": "Write a function to find the nth Fibonacci number.",
"domain": "code"
}
]
# Run tests with both CoD and CoT
for approach in ["CoD", "CoT"]:
print(f"\n===== Testing with {approach} approach =====\n")
for test in test_problems:
print(f"Problem ({test['domain']}): {test['problem']}")
# Solve with specified approach
result = await client.solve_with_reasoning(
problem=test['problem'],
domain=test['domain'],
approach=approach
)
print(f"\nReasoning:\n{result['reasoning_steps']}")
print(f"\nAnswer: {result['final_answer']}")
print(f"Tokens: {result['token_count']}")
print(f"Word limit: {result['word_limit']}")
print(f"Complexity: {result['complexity']}")
print("\n" + "-" * 50 + "\n")
# Get performance stats
print("\n===== Performance Statistics =====\n")
stats = await client.get_performance_stats()
for stat in stats:
print(f"Domain: {stat['domain']}")
print(f"Approach: {stat['approach']}")
print(f"Average tokens: {stat['avg_tokens']}")
print(f"Average time: {stat['avg_time_ms']}")
if stat['accuracy'] is not None:
print(f"Accuracy: {stat['accuracy'] * 100:.1f}%")
print(f"Sample size: {stat['count']}")
print()
# Get token reduction stats
reduction_stats = await client.get_token_reduction_stats()
if reduction_stats:
print("\n===== Token Reduction =====\n")
for stat in reduction_stats:
print(f"Domain: {stat['domain']}")
print(f"CoD avg tokens: {stat['cod_avg_tokens']:.1f}")
print(f"CoT avg tokens: {stat['cot_avg_tokens']:.1f}")
print(f"Reduction: {stat['reduction_percentage']:.1f}%")
print()
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
name: '@brendancopley/mcp-chain-of-draft-prompt-tool'
description: |-
A TypeScript-based Model Context Protocol (MCP) server for Chain of Draft
reasoning
version: "1.0.0"
build:
dockerBuildPath: ./
startCommand:
type: stdio
configSchema:
type: object
title: MCP Server Configuration
description: Configuration options for the MCP server
properties:
LLM_PROVIDER:
type: string
description: Choose which provider to use - 'anthropic', 'openai', 'mistral', or 'ollama'
enum: ['anthropic', 'openai', 'mistral', 'ollama']
default: 'anthropic'
LLM_MODEL:
type: string
description: Default model to use (provider-specific)
default: 'claude-3-7-sonnet-latest'
ANTHROPIC_API_KEY:
type: string
description: API key for Anthropic Claude models
ANTHROPIC_BASE_URL:
type: string
description: Base URL for Anthropic API
default: 'https://api.anthropic.com'
OPENAI_API_KEY:
type: string
description: API key for OpenAI models
OPENAI_BASE_URL:
type: string
description: Base URL for OpenAI API
default: 'https://api.openai.com'
MISTRAL_API_KEY:
type: string
description: API key for Mistral AI models
OLLAMA_BASE_URL:
type: string
description: Base URL for Ollama local deployment
default: 'http://localhost:11434'
COD_MAX_WORDS_PER_STEP:
type: string
description: Maximum words per step in Chain of Draft
default: '5'
ENFORCE_FORMAT:
type: string
description: Whether to enforce format in Chain of Draft
default: 'true'
ADAPTIVE_WORD_LIMIT:
type: string
description: Whether to use adaptive word limit in Chain of Draft
default: 'true'
COD_DB_URL:
type: string
description: URL for Chain of Draft analytics database
default: 'sqlite:///cod_analytics.db'
COD_EXAMPLES_DB:
type: string
description: Path to Chain of Draft examples database
default: 'cod_examples.db'
COD_DEFAULT_MODEL:
type: string
description: Default model for Chain of Draft
default: 'claude-3-7-sonnet-latest'
COD_MAX_TOKENS:
type: string
description: Maximum tokens for Chain of Draft
default: '500'
required: []
commandFunction: |-
(config) => ({
command: 'node',
args: ['dist/index.js'],
env: {
NODE_ENV: config.NODE_ENV || 'production',
LLM_PROVIDER: config.LLM_PROVIDER || 'anthropic',
LLM_MODEL: config.LLM_MODEL || 'claude-3-7-sonnet-latest',
ANTHROPIC_API_KEY: config.ANTHROPIC_API_KEY,
OPENAI_API_KEY: config.OPENAI_API_KEY,
MISTRAL_API_KEY: config.MISTRAL_API_KEY,
ANTHROPIC_BASE_URL: config.ANTHROPIC_BASE_URL || 'https://api.anthropic.com',
OPENAI_BASE_URL: config.OPENAI_BASE_URL || 'https://api.openai.com',
OLLAMA_BASE_URL: config.OLLAMA_BASE_URL || 'http://localhost:11434',
COD_MAX_WORDS_PER_STEP: config.COD_MAX_WORDS_PER_STEP || '5',
ENFORCE_FORMAT: config.ENFORCE_FORMAT || 'true',
ADAPTIVE_WORD_LIMIT: config.ADAPTIVE_WORD_LIMIT || 'true',
COD_DB_URL: config.COD_DB_URL || 'sqlite:///cod_analytics.db',
COD_EXAMPLES_DB: config.COD_EXAMPLES_DB || 'cod_examples.db',
COD_DEFAULT_MODEL: config.COD_DEFAULT_MODEL || 'claude-3-7-sonnet-latest',
COD_MAX_TOKENS: config.COD_MAX_TOKENS || '500'
}
})
clients:
- claude
- cursor
- windsurf
- cline
- typescript
```
--------------------------------------------------------------------------------
/promptsConfig.json:
--------------------------------------------------------------------------------
```json
{
"categories": [
{
"id": "chain-of-draft",
"name": "Chain of Draft",
"description": "Prompts related to Chain of Draft reasoning approach"
},
{
"id": "analytics",
"name": "Analytics",
"description": "Tools for analyzing performance and complexity"
}
],
"prompts": [
{
"id": "chain_of_draft_solve",
"name": "Chain of Draft Solve",
"category": "chain-of-draft",
"description": "Solve a reasoning problem using Chain of Draft approach",
"arguments": [
{
"name": "problem",
"description": "The problem to solve",
"required": true
},
{
"name": "domain",
"description": "Domain for context (math, logic, code, common-sense, etc.)",
"required": false
},
{
"name": "max_words_per_step",
"description": "Maximum words per reasoning step (default: adaptive)",
"required": false
},
{
"name": "approach",
"description": "Force 'CoD' or 'CoT' approach (default: auto-select)",
"required": false
},
{
"name": "enforce_format",
"description": "Whether to enforce the word limit (default: True)",
"required": false
},
{
"name": "adaptive_word_limit",
"description": "Adjust word limits based on complexity (default: True)",
"required": false
}
]
},
{
"id": "math_solve",
"name": "Math Problem Solve",
"category": "chain-of-draft",
"description": "Solve a math problem using Chain of Draft reasoning",
"arguments": [
{
"name": "problem",
"description": "The math problem to solve",
"required": true
},
{
"name": "approach",
"description": "Force 'CoD' or 'CoT' approach (default: auto-select)",
"required": false
},
{
"name": "max_words_per_step",
"description": "Maximum words per step (default: adaptive)",
"required": false
}
]
},
{
"id": "code_solve",
"name": "Code Problem Solve",
"category": "chain-of-draft",
"description": "Solve a coding problem using Chain of Draft reasoning",
"arguments": [
{
"name": "problem",
"description": "The coding problem to solve",
"required": true
},
{
"name": "approach",
"description": "Force 'CoD' or 'CoT' approach (default: auto-select)",
"required": false
},
{
"name": "max_words_per_step",
"description": "Maximum words per step (default: adaptive)",
"required": false
}
]
},
{
"id": "logic_solve",
"name": "Logic Problem Solve",
"category": "chain-of-draft",
"description": "Solve a logic problem using Chain of Draft reasoning",
"arguments": [
{
"name": "problem",
"description": "The logic problem to solve",
"required": true
},
{
"name": "approach",
"description": "Force 'CoD' or 'CoT' approach (default: auto-select)",
"required": false
},
{
"name": "max_words_per_step",
"description": "Maximum words per step (default: adaptive)",
"required": false
}
]
},
{
"id": "get_performance_stats",
"name": "Get Performance Stats",
"category": "analytics",
"description": "Get performance statistics for CoD vs CoT approaches",
"arguments": [
{
"name": "domain",
"description": "Filter for specific domain (optional)",
"required": false
}
]
},
{
"id": "get_token_reduction",
"name": "Get Token Reduction",
"category": "analytics",
"description": "Get token reduction statistics for CoD vs CoT",
"arguments": []
},
{
"id": "analyze_problem_complexity",
"name": "Analyze Problem Complexity",
"category": "analytics",
"description": "Analyze the complexity of a problem",
"arguments": [
{
"name": "problem",
"description": "The problem to analyze",
"required": true
},
{
"name": "domain",
"description": "Problem domain",
"required": false
}
]
}
],
"imports": []
}
```
--------------------------------------------------------------------------------
/src/server.ts:
--------------------------------------------------------------------------------
```typescript
#!/usr/bin/env node
/**
* Node.js wrapper for the Chain of Draft Python server
* This provides better compatibility with Claude Desktop
*/
import { spawn } from 'child_process';
import path from 'path';
import fs from 'fs';
import { fileURLToPath } from 'url';
import { logger } from './utils/logger.js';
import dotenv from 'dotenv';
// Load environment variables
dotenv.config();
// Get current file directory in ESM
const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);
// Path to the Python server script
const serverPath = path.join(__dirname, 'server.py');
logger.devLog('Server initialization', {
serverPath,
nodeEnv: process.env.NODE_ENV,
cwd: process.cwd(),
pythonPath: process.env.PYTHON_PATH || 'python3'
});
// Error if server.py doesn't exist
if (!fs.existsSync(serverPath)) {
logger.error(`Server file not found at ${serverPath}`);
process.exit(1);
}
// Launch the Python process with debugging
logger.debug(`Starting Python process: python3 ${serverPath}`);
const pythonProcess = spawn('python3', [serverPath], {
env: {
...process.env,
PYTHONUNBUFFERED: '1', // Ensure Python output isn't buffered
DEBUG: process.env.NODE_ENV === 'development' ? '1' : '0'
},
stdio: ['pipe', 'pipe', 'pipe'] // Explicitly define stdio
});
logger.devLog('Python process spawned', {
pid: pythonProcess.pid,
spawnargs: pythonProcess.spawnargs,
env: {
PYTHONUNBUFFERED: process.env.PYTHONUNBUFFERED,
DEBUG: process.env.DEBUG
}
});
// Keep the process alive by sending a dummy input occasionally
const keepAlive = setInterval(() => {
// Just checking if process is still running
if (pythonProcess.killed) {
clearInterval(keepAlive);
logger.error("Python process was killed, exiting");
logger.devLog('Python process killed', {
pid: pythonProcess.pid,
exitCode: pythonProcess.exitCode,
signalCode: pythonProcess.signalCode
});
process.exit(1);
}
}, 5000);
// Pass stdin to the Python process
process.stdin.pipe(pythonProcess.stdin);
// Pipe Python's stdout to our stdout
pythonProcess.stdout.pipe(process.stdout);
// Log stderr but don't pipe it to avoid protocol errors
pythonProcess.stderr.on('data', (data: Buffer) => {
process.stderr.write(data);
if (process.env.NODE_ENV === 'development') {
logger.devLog('Python stderr', data.toString());
}
});
// Handle process termination
pythonProcess.on('close', (code: number | null) => {
if (code === null) {
logger.error('Python process was killed, exiting');
logger.devLog('Python process terminated', {
pid: pythonProcess.pid,
exitCode: pythonProcess.exitCode,
signalCode: pythonProcess.signalCode
});
} else {
logger.error(`Chain of Draft server exited with code ${code}`);
logger.devLog('Python process exited', {
pid: pythonProcess.pid,
code,
signalCode: pythonProcess.signalCode
});
}
process.exit(code || 1);
});
// Forward termination signals
process.on('SIGINT', () => {
logger.devLog('Received SIGINT signal');
pythonProcess.kill('SIGINT');
});
process.on('SIGTERM', () => {
logger.devLog('Received SIGTERM signal');
pythonProcess.kill('SIGTERM');
});
export async function startServer() {
const serverPath = path.join(process.cwd(), 'server.py');
logger.devLog('Starting server', {
serverPath,
nodeEnv: process.env.NODE_ENV,
development: process.env.NODE_ENV === 'development'
});
if (!fs.existsSync(serverPath)) {
logger.error(`Server file not found at ${serverPath}`);
process.exit(1);
}
logger.debug(`Starting Python process: python3 ${serverPath}`);
const pythonProcess = spawn('python3', [serverPath], {
stdio: ['pipe', 'pipe', 'pipe'],
env: {
...process.env,
PYTHONUNBUFFERED: '1',
DEBUG: process.env.NODE_ENV === 'development' ? '1' : '0'
}
});
logger.devLog('Server process spawned', {
pid: pythonProcess.pid,
spawnargs: pythonProcess.spawnargs
});
pythonProcess.stdout.on('data', (data) => {
process.stdout.write(data);
if (process.env.NODE_ENV === 'development') {
logger.devLog('Python stdout', data.toString());
}
});
pythonProcess.stderr.on('data', (data) => {
process.stderr.write(data);
if (process.env.NODE_ENV === 'development') {
logger.devLog('Python stderr', data.toString());
}
});
pythonProcess.on('close', (code) => {
if (code === null) {
logger.error('Python process was killed, exiting');
logger.devLog('Server process killed', {
pid: pythonProcess.pid,
exitCode: pythonProcess.exitCode
});
} else {
logger.error(`Chain of Draft server exited with code ${code}`);
logger.devLog('Server process exited', {
pid: pythonProcess.pid,
code
});
}
process.exit(code || 1);
});
return pythonProcess;
}
```
--------------------------------------------------------------------------------
/src/python/format.py:
--------------------------------------------------------------------------------
```python
"""
Format enforcement module for the Chain of Draft MCP server.
Ensures reasoning steps adhere to the word limit.
"""
import re
class FormatEnforcer:
"""
Enforces the Chain of Draft format by ensuring reasoning steps
adhere to the specified word limit.
"""
def __init__(self):
"""Initialize with patterns for detecting reasoning steps."""
# Patterns for identifying different step formats
self.step_pattern = re.compile(
r'(\d+\.\s*|Step\s+\d+:|\n-\s+|\n\*\s+|•\s+|^\s*-\s+|^\s*\*\s+)',
re.MULTILINE
)
def enforce_word_limit(self, reasoning, max_words_per_step):
"""
Enforce word limit per reasoning step.
Args:
reasoning: The reasoning text to process
max_words_per_step: Maximum number of words allowed per step
Returns:
Processed reasoning with enforced word limits
"""
# Split into steps
steps = self._split_into_steps(reasoning)
# Process each step
enforced_steps = []
for step in steps:
enforced_step = self._enforce_step(step, max_words_per_step)
enforced_steps.append(enforced_step)
# Combine back
return "\n".join(enforced_steps)
def _split_into_steps(self, reasoning):
"""Split reasoning text into individual steps."""
# Try to detect step formatting
if self.step_pattern.search(reasoning):
# Extract steps with their markers
parts = []
current_part = ""
lines = reasoning.split('\n')
for line in lines:
# If this is a new step, store the previous part and start a new one
if self.step_pattern.match(line) or re.match(r'^\s*\d+\.', line):
if current_part:
parts.append(current_part)
current_part = line
else:
# Otherwise add to current part
if current_part:
current_part += "\n" + line
else:
current_part = line
# Add the last part
if current_part:
parts.append(current_part)
return parts if parts else [reasoning]
else:
# If no clear step markers, split by lines
return [line.strip() for line in reasoning.split('\n') if line.strip()]
def _enforce_step(self, step, max_words):
"""Enforce word limit on a single reasoning step."""
words = step.split()
# If already within limit, return as is
if len(words) <= max_words:
return step
# Extract step number/marker if present
match = re.match(r'^(\d+\.\s*|Step\s+\d+:|\s*-\s+|\s*\*\s+|•\s+)', step)
marker = match.group(0) if match else ""
content = step[len(marker):].strip() if marker else step
# Truncate content words
content_words = content.split()
truncated = " ".join(content_words[:max_words])
# Reassemble with marker
return f"{marker}{truncated}"
def analyze_adherence(self, reasoning, max_words_per_step):
"""
Analyze how well the reasoning adheres to word limits.
Args:
reasoning: The reasoning text to analyze
max_words_per_step: Maximum number of words allowed per step
Returns:
Dictionary with adherence metrics
"""
steps = self._split_into_steps(reasoning)
# Count words in each step, excluding markers
step_counts = []
for step in steps:
# Extract step number/marker if present
match = re.match(r'^(\d+\.\s*|Step\s+\d+:|\s*-\s+|\s*\*\s+|•\s+)', step)
marker = match.group(0) if match else ""
content = step[len(marker):].strip() if marker else step
# Count words in content
step_counts.append(len(content.split()))
# Calculate adherence metrics
adherence = {
"total_steps": len(steps),
"steps_within_limit": sum(1 for count in step_counts if count <= max_words_per_step),
"average_words_per_step": sum(step_counts) / len(steps) if steps else 0,
"max_words_in_any_step": max(step_counts) if steps else 0,
"adherence_rate": sum(1 for count in step_counts if count <= max_words_per_step) / len(steps) if steps else 1.0,
"step_counts": step_counts
}
return adherence
def format_to_numbered_steps(self, reasoning):
"""Format reasoning into consistently numbered steps."""
steps = self._split_into_steps(reasoning)
# Convert to numbered steps
numbered_steps = []
for i, step in enumerate(steps, 1):
# Remove any existing markers
match = re.match(r'^(\d+\.\s*|Step\s+\d+:|\s*-\s+|\s*\*\s+|•\s+)', step)
if match:
content = step[len(match.group(0)):].strip()
else:
content = step.strip()
# Add numbered step format
numbered_steps.append(f"{i}. {content}")
return "\n".join(numbered_steps)
```
--------------------------------------------------------------------------------
/src/python/complexity.py:
--------------------------------------------------------------------------------
```python
"""
Complexity estimation module for the Chain of Draft MCP server.
Analyzes problems to determine appropriate word limits.
"""
class ComplexityEstimator:
"""
Estimates the complexity of a problem to determine appropriate word limits
for Chain of Draft reasoning steps.
"""
def __init__(self):
"""Initialize with domain-specific complexity rules."""
# Base complexity rules - default word limits per domain
self.domain_base_limits = {
"math": 6,
"logic": 5,
"common_sense": 4,
"physics": 7,
"chemistry": 6,
"biology": 5,
"code": 8,
"puzzle": 5,
"general": 5
}
# Keywords that might indicate complexity
self.complexity_indicators = {
"math": [
"integral", "derivative", "equation", "proof", "theorem", "calculus",
"matrix", "vector", "linear algebra", "probability", "statistics",
"geometric series", "differential", "polynomial", "factorial"
],
"logic": [
"if and only if", "necessary condition", "sufficient", "contradiction",
"syllogism", "premise", "fallacy", "converse", "counterexample",
"logical equivalence", "negation", "disjunction", "conjunction"
],
"code": [
"recursion", "algorithm", "complexity", "optimization", "function",
"class", "object", "inheritance", "polymorphism", "data structure",
"binary tree", "hash table", "graph", "dynamic programming"
],
"physics": [
"quantum", "relativity", "momentum", "force", "acceleration",
"energy", "thermodynamics", "electric field", "magnetic field",
"potential", "entropy", "wavelength", "frequency"
],
"chemistry": [
"reaction", "molecule", "compound", "element", "equilibrium",
"acid", "base", "oxidation", "reduction", "catalyst", "isomer"
],
"biology": [
"gene", "protein", "enzyme", "cell", "tissue", "organ", "system",
"metabolism", "photosynthesis", "respiration", "homeostasis"
],
"puzzle": [
"constraint", "sequence", "pattern", "rules", "probability",
"combination", "permutation", "optimal", "strategy"
]
}
async def estimate_complexity(self, problem, domain="general"):
"""
Estimate the complexity of a problem based on its characteristics.
Returns a recommended word limit per step.
"""
# Start with base word limit for domain
base_limit = self.domain_base_limits.get(domain.lower(), 5)
# Analyze problem length - longer problems often need more detailed reasoning
length_factor = min(len(problem.split()) / 50, 2) # Cap at doubling
# Check for complexity indicators
indicator_count = 0
indicators = self.complexity_indicators.get(domain.lower(), [])
for indicator in indicators:
if indicator.lower() in problem.lower():
indicator_count += 1
indicator_factor = min(1 + (indicator_count * 0.2), 1.8) # Cap at 80% increase
# Count the number of question marks - might indicate multi-part problems
question_factor = 1 + (problem.count("?") * 0.2)
# Simple complexity analysis based on sentence structure
sentences = [s for s in problem.split(".") if s.strip()]
words_per_sentence = len(problem.split()) / max(len(sentences), 1)
sentence_complexity_factor = min(words_per_sentence / 15, 1.5) # Complex sentences
# Special domain-specific adjustments
domain_factor = 1.0
if domain.lower() == "math" and any(term in problem.lower() for term in ["prove", "proof", "theorem"]):
domain_factor = 1.3 # Proofs need more explanation
elif domain.lower() == "code" and any(term in problem.lower() for term in ["implement", "function", "algorithm"]):
domain_factor = 1.2 # Code implementations need more detail
# Combine factors - take the maximum impact factor
impact_factor = max(length_factor, indicator_factor, question_factor,
sentence_complexity_factor, domain_factor)
# Apply the impact factor to the base limit
adjusted_limit = round(base_limit * impact_factor)
# Cap at reasonable bounds
return max(3, min(adjusted_limit, 10))
def analyze_problem(self, problem, domain="general"):
"""
Provide a detailed analysis of problem complexity factors.
Useful for debugging and understanding complexity estimates.
"""
base_limit = self.domain_base_limits.get(domain.lower(), 5)
# Word count analysis
word_count = len(problem.split())
length_factor = min(word_count / 50, 2)
# Indicator analysis
indicators = self.complexity_indicators.get(domain.lower(), [])
found_indicators = [ind for ind in indicators if ind.lower() in problem.lower()]
indicator_count = len(found_indicators)
indicator_factor = min(1 + (indicator_count * 0.2), 1.8)
# Question mark analysis
question_count = problem.count("?")
question_factor = 1 + (question_count * 0.2)
# Sentence complexity
sentences = [s for s in problem.split(".") if s.strip()]
words_per_sentence = word_count / max(len(sentences), 1)
sentence_complexity_factor = min(words_per_sentence / 15, 1.5)
return {
"domain": domain,
"base_limit": base_limit,
"word_count": word_count,
"length_factor": length_factor,
"indicator_count": indicator_count,
"found_indicators": found_indicators,
"indicator_factor": indicator_factor,
"question_count": question_count,
"question_factor": question_factor,
"sentence_count": len(sentences),
"words_per_sentence": words_per_sentence,
"sentence_complexity_factor": sentence_complexity_factor,
"estimated_complexity": max(3, min(round(base_limit * max(length_factor, indicator_factor, question_factor, sentence_complexity_factor)), 10))
}
```
--------------------------------------------------------------------------------
/src/python/reasoning.py:
--------------------------------------------------------------------------------
```python
"""
Reasoning selector module for the Chain of Draft MCP server.
Handles choosing between CoD and CoT approaches.
"""
from complexity import ComplexityEstimator
class ReasoningSelector:
"""
Selects the most appropriate reasoning approach (CoD or CoT)
based on problem characteristics and historical performance.
"""
def __init__(self, analytics_service):
"""Initialize with analytics service for performance data."""
self.analytics = analytics_service
# Preferences for when to use which approach
self.default_preferences = {
# Format: domain -> criteria
"math": {
"complexity_threshold": 7, # Use CoT for problems with complexity above this
"accuracy_threshold": 0.85 # Use CoT if CoD accuracy falls below this
},
"code": {
"complexity_threshold": 8,
"accuracy_threshold": 0.9
},
"physics": {
"complexity_threshold": 7,
"accuracy_threshold": 0.85
},
"chemistry": {
"complexity_threshold": 7,
"accuracy_threshold": 0.85
},
"biology": {
"complexity_threshold": 6,
"accuracy_threshold": 0.85
},
"logic": {
"complexity_threshold": 6,
"accuracy_threshold": 0.9
},
"puzzle": {
"complexity_threshold": 7,
"accuracy_threshold": 0.85
},
# Default for any domain
"default": {
"complexity_threshold": 6,
"accuracy_threshold": 0.8
}
}
async def select_approach(self, problem, domain, complexity_score=None):
"""
Select whether to use CoD or CoT for this problem.
Args:
problem: The problem text
domain: Problem domain (math, code, logic, etc.)
complexity_score: Pre-computed complexity score (optional)
Returns:
Tuple of (approach, reason) where approach is "CoD" or "CoT"
"""
# Get domain-specific preferences
prefs = self.default_preferences.get(domain.lower(), self.default_preferences["default"])
# If complexity score not provided, estimate it
if complexity_score is None:
estimator = ComplexityEstimator()
complexity_score = await estimator.estimate_complexity(problem, domain)
# Check if complexity exceeds threshold
if complexity_score > prefs["complexity_threshold"]:
return "CoT", f"Problem complexity ({complexity_score}) exceeds threshold ({prefs['complexity_threshold']})"
# Check historical accuracy for this domain with CoD
domain_performance = await self.analytics.get_performance_by_domain(domain)
cod_accuracy = next((p["accuracy"] for p in domain_performance
if p["approach"] == "CoD"), None)
if cod_accuracy is not None and cod_accuracy < prefs["accuracy_threshold"]:
return "CoT", f"Historical accuracy with CoD ({cod_accuracy:.2f}) below threshold ({prefs['accuracy_threshold']})"
# Default to CoD for efficiency
return "CoD", "Default to Chain-of-Draft for efficiency"
def update_preferences(self, domain, complexity_threshold=None, accuracy_threshold=None):
"""Update preferences for a specific domain."""
if domain not in self.default_preferences:
self.default_preferences[domain] = self.default_preferences["default"].copy()
if complexity_threshold is not None:
self.default_preferences[domain]["complexity_threshold"] = complexity_threshold
if accuracy_threshold is not None:
self.default_preferences[domain]["accuracy_threshold"] = accuracy_threshold
def get_preferences(self, domain=None):
"""Get current preferences for all domains or a specific domain."""
if domain:
return self.default_preferences.get(domain, self.default_preferences["default"])
return self.default_preferences
def create_cod_prompt(problem, domain, max_words_per_step, examples=None):
"""
Create a Chain of Draft prompt for the LLM.
Args:
problem: The problem to solve
domain: Domain for context (math, logic, common-sense, etc.)
max_words_per_step: Maximum words per reasoning step
examples: Optional list of few-shot examples
Returns:
Dictionary with system and user prompts
"""
system_prompt = f"""
You are an expert problem solver using Chain of Draft reasoning.
Think step by step, but only keep a minimum draft for each thinking step,
with {max_words_per_step} words at most per step.
Return the answer at the end after '####'.
"""
# Add domain-specific context
if domain.lower() == "math":
system_prompt += "\nUse mathematical notation to keep steps concise."
elif domain.lower() == "code":
system_prompt += "\nUse pseudocode or short code snippets when appropriate."
elif domain.lower() == "physics":
system_prompt += "\nUse equations and physical quantities with units."
# Add examples if provided
example_text = ""
if examples:
for example in examples:
example_text += f"\nProblem: {example['problem']}\nSolution:\n{example['reasoning']}\n####\n{example['answer']}\n"
user_prompt = f"Problem: {problem}"
return {
"system": system_prompt,
"user": example_text + "\n" + user_prompt if example_text else user_prompt
}
def create_cot_prompt(problem, domain, examples=None):
"""
Create a Chain of Thought prompt for the LLM.
Args:
problem: The problem to solve
domain: Domain for context (math, logic, common-sense, etc.)
examples: Optional list of few-shot examples
Returns:
Dictionary with system and user prompts
"""
system_prompt = """
Think step by step to answer the following question.
Return the answer at the end of the response after a separator ####.
"""
# Add domain-specific context
if domain.lower() == "math":
system_prompt += "\nMake sure to show all mathematical operations clearly."
elif domain.lower() == "code":
system_prompt += "\nBe detailed about algorithms and implementation steps."
elif domain.lower() == "physics":
system_prompt += "\nExplain physical principles and equations in detail."
# Add examples if provided
example_text = ""
if examples:
for example in examples:
example_text += f"\nProblem: {example['problem']}\nSolution:\n{example['reasoning']}\n####\n{example['answer']}\n"
user_prompt = f"Problem: {problem}"
return {
"system": system_prompt,
"user": example_text + "\n" + user_prompt if example_text else user_prompt
}
```
--------------------------------------------------------------------------------
/src/python/analytics.py:
--------------------------------------------------------------------------------
```python
"""
Analytics service for the Chain of Draft MCP server.
Tracks performance metrics for different reasoning approaches.
"""
import datetime
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, JSON, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
Base = declarative_base()
class InferenceRecord(Base):
"""Database model for tracking inference performance."""
__tablename__ = 'inference_records'
id = Column(Integer, primary_key=True)
timestamp = Column(DateTime, default=datetime.datetime.utcnow)
problem_id = Column(String)
problem_text = Column(String)
domain = Column(String)
approach = Column(String) # "CoD" or "CoT"
word_limit = Column(Integer)
tokens_used = Column(Integer)
execution_time_ms = Column(Float)
reasoning_steps = Column(String)
answer = Column(String)
expected_answer = Column(String, nullable=True)
is_correct = Column(Integer, nullable=True) # 1=correct, 0=incorrect, null=unknown
meta_data = Column(JSON, nullable=True) # Changed from metadata to meta_data to avoid SQLAlchemy reserved keyword
class AnalyticsService:
"""Service for tracking and analyzing inference performance."""
def __init__(self, db_url=None):
"""Initialize the analytics service with a database connection."""
if db_url is None:
# Default to SQLite in the current directory
db_url = os.environ.get("COD_DB_URL", "sqlite:///cod_analytics.db")
self.engine = create_engine(db_url)
Base.metadata.create_all(self.engine)
self.Session = sessionmaker(bind=self.engine)
async def record_inference(self, problem, domain, approach, word_limit,
tokens_used, execution_time, reasoning, answer,
expected_answer=None, metadata=None):
"""Record a new inference with performance metrics."""
session = self.Session()
try:
# Simple hash function for problem ID
problem_id = str(abs(hash(problem)) % (10 ** 10))
record = InferenceRecord(
problem_id=problem_id,
problem_text=problem,
domain=domain,
approach=approach,
word_limit=word_limit,
tokens_used=tokens_used,
execution_time_ms=execution_time,
reasoning_steps=reasoning,
answer=answer,
expected_answer=expected_answer,
is_correct=self._check_correctness(answer, expected_answer) if expected_answer else None,
meta_data=metadata
)
session.add(record)
session.commit()
return record.id
finally:
session.close()
def _check_correctness(self, answer, expected_answer):
"""Check if an answer is correct."""
# Basic string comparison - could be improved with more sophisticated matching
if not answer or not expected_answer:
return None
return 1 if answer.strip().lower() == expected_answer.strip().lower() else 0
async def get_performance_by_domain(self, domain=None):
"""Get performance statistics by domain."""
session = self.Session()
try:
query = session.query(
InferenceRecord.domain,
InferenceRecord.approach,
func.avg(InferenceRecord.tokens_used).label("avg_tokens"),
func.avg(InferenceRecord.execution_time_ms).label("avg_time"),
func.avg(InferenceRecord.is_correct).label("accuracy"),
func.count(InferenceRecord.id).label("count")
).group_by(InferenceRecord.domain, InferenceRecord.approach)
if domain:
query = query.filter(InferenceRecord.domain == domain)
results = query.all()
return [
{
"domain": r.domain,
"approach": r.approach,
"avg_tokens": r.avg_tokens,
"avg_time_ms": r.avg_time,
"accuracy": r.accuracy if r.accuracy is not None else None,
"count": r.count
}
for r in results
]
finally:
session.close()
async def get_token_reduction_stats(self):
"""Calculate token reduction statistics for CoD vs CoT."""
session = self.Session()
try:
domains = session.query(InferenceRecord.domain).distinct().all()
results = []
for domain_row in domains:
domain = domain_row[0]
# Get average tokens for CoD and CoT approaches in this domain
cod_avg = session.query(func.avg(InferenceRecord.tokens_used)).filter(
InferenceRecord.domain == domain,
InferenceRecord.approach == "CoD"
).scalar() or 0
cot_avg = session.query(func.avg(InferenceRecord.tokens_used)).filter(
InferenceRecord.domain == domain,
InferenceRecord.approach == "CoT"
).scalar() or 0
if cot_avg > 0:
reduction_percentage = (1 - (cod_avg / cot_avg)) * 100
else:
reduction_percentage = 0
results.append({
"domain": domain,
"cod_avg_tokens": cod_avg,
"cot_avg_tokens": cot_avg,
"reduction_percentage": reduction_percentage
})
return results
finally:
session.close()
async def get_accuracy_comparison(self):
"""Compare accuracy between CoD and CoT approaches."""
session = self.Session()
try:
domains = session.query(InferenceRecord.domain).distinct().all()
results = []
for domain_row in domains:
domain = domain_row[0]
# Get accuracy for CoD and CoT approaches in this domain
cod_accuracy = session.query(func.avg(InferenceRecord.is_correct)).filter(
InferenceRecord.domain == domain,
InferenceRecord.approach == "CoD",
InferenceRecord.is_correct.isnot(None)
).scalar()
cot_accuracy = session.query(func.avg(InferenceRecord.is_correct)).filter(
InferenceRecord.domain == domain,
InferenceRecord.approach == "CoT",
InferenceRecord.is_correct.isnot(None)
).scalar()
results.append({
"domain": domain,
"cod_accuracy": cod_accuracy,
"cot_accuracy": cot_accuracy,
"accuracy_difference": (cod_accuracy - cot_accuracy) if cod_accuracy and cot_accuracy else None
})
return results
finally:
session.close()
```
--------------------------------------------------------------------------------
/src/python/server.py:
--------------------------------------------------------------------------------
```python
"""
Main server module for the Chain of Draft MCP server.
Implements the MCP server with reasoning tools.
"""
import os
import asyncio
import time
from dotenv import load_dotenv
# Import the FastMCP module for modern MCP API
from mcp.server.fastmcp import FastMCP
from mcp.server.stdio import stdio_server
# Import other modules
from analytics import AnalyticsService
from complexity import ComplexityEstimator
from examples import ExampleDatabase
from format import FormatEnforcer
from reasoning import ReasoningSelector, create_cod_prompt, create_cot_prompt
from client import ChainOfDraftClient
# Load environment variables
load_dotenv()
# Initialize services
analytics = AnalyticsService()
complexity_estimator = ComplexityEstimator()
example_db = ExampleDatabase()
format_enforcer = FormatEnforcer()
cod_client = ChainOfDraftClient()
# Initialize FastMCP server
app = FastMCP("mcp-chain-of-draft-prompt-tool")
@app.tool()
async def chain_of_draft_solve(
problem: str,
domain: str = "general",
max_words_per_step: int = None,
approach: str = None,
enforce_format: bool = True,
adaptive_word_limit: bool = True
) -> str:
"""Solve a reasoning problem using Chain of Draft approach.
Args:
problem: The problem to solve
domain: Domain for context (math, logic, code, common-sense, etc.)
max_words_per_step: Maximum words per reasoning step (default: adaptive)
approach: Force "CoD" or "CoT" approach (default: auto-select)
enforce_format: Whether to enforce the word limit (default: True)
adaptive_word_limit: Adjust word limits based on complexity (default: True)
"""
# Track execution time
start_time = time.time()
# Process the request with the client
result = await cod_client.solve_with_reasoning(
problem=problem,
domain=domain,
max_words_per_step=max_words_per_step,
approach=approach,
enforce_format=enforce_format,
adaptive_word_limit=adaptive_word_limit
)
# Calculate execution time
execution_time = (time.time() - start_time) * 1000 # ms
# Format the response
formatted_response = (
f"Chain of {result['approach']} reasoning ({result['word_limit']} word limit):\n\n"
f"{result['reasoning_steps']}\n\n"
f"Final answer: {result['final_answer']}\n\n"
f"Stats: {result['token_count']} tokens, {execution_time:.0f}ms, "
f"complexity score: {result['complexity']}"
)
return formatted_response
@app.tool()
async def math_solve(
problem: str,
approach: str = None,
max_words_per_step: int = None
) -> str:
"""Solve a math problem using Chain of Draft reasoning.
Args:
problem: The math problem to solve
approach: Force "CoD" or "CoT" approach (default: auto-select)
max_words_per_step: Maximum words per step (default: adaptive)
"""
return await chain_of_draft_solve(
problem=problem,
domain="math",
approach=approach,
max_words_per_step=max_words_per_step
)
@app.tool()
async def code_solve(
problem: str,
approach: str = None,
max_words_per_step: int = None
) -> str:
"""Solve a coding problem using Chain of Draft reasoning.
Args:
problem: The coding problem to solve
approach: Force "CoD" or "CoT" approach (default: auto-select)
max_words_per_step: Maximum words per step (default: adaptive)
"""
return await chain_of_draft_solve(
problem=problem,
domain="code",
approach=approach,
max_words_per_step=max_words_per_step
)
@app.tool()
async def logic_solve(
problem: str,
approach: str = None,
max_words_per_step: int = None
) -> str:
"""Solve a logic problem using Chain of Draft reasoning.
Args:
problem: The logic problem to solve
approach: Force "CoD" or "CoT" approach (default: auto-select)
max_words_per_step: Maximum words per step (default: adaptive)
"""
return await chain_of_draft_solve(
problem=problem,
domain="logic",
approach=approach,
max_words_per_step=max_words_per_step
)
@app.tool()
async def get_performance_stats(
domain: str = None
) -> str:
"""Get performance statistics for CoD vs CoT approaches.
Args:
domain: Filter for specific domain (optional)
"""
stats = await analytics.get_performance_by_domain(domain)
result = "Performance Comparison (CoD vs CoT):\n\n"
if not stats:
return "No performance data available yet."
for stat in stats:
result += f"Domain: {stat['domain']}\n"
result += f"Approach: {stat['approach']}\n"
result += f"Average tokens: {stat['avg_tokens']:.1f}\n"
result += f"Average time: {stat['avg_time_ms']:.1f}ms\n"
if stat['accuracy'] is not None:
result += f"Accuracy: {stat['accuracy'] * 100:.1f}%\n"
result += f"Sample size: {stat['count']}\n\n"
return result
@app.tool()
async def get_token_reduction() -> str:
"""Get token reduction statistics for CoD vs CoT."""
stats = await analytics.get_token_reduction_stats()
result = "Token Reduction Analysis:\n\n"
if not stats:
return "No reduction data available yet."
for stat in stats:
result += f"Domain: {stat['domain']}\n"
result += f"CoD avg tokens: {stat['cod_avg_tokens']:.1f}\n"
result += f"CoT avg tokens: {stat['cot_avg_tokens']:.1f}\n"
result += f"Reduction: {stat['reduction_percentage']:.1f}%\n\n"
return result
@app.tool()
async def analyze_problem_complexity(
problem: str,
domain: str = "general"
) -> str:
"""Analyze the complexity of a problem.
Args:
problem: The problem to analyze
domain: Problem domain
"""
analysis = complexity_estimator.analyze_problem(problem, domain)
result = f"Complexity Analysis for {domain} problem:\n\n"
result += f"Word count: {analysis['word_count']}\n"
result += f"Sentence count: {analysis['sentence_count']}\n"
result += f"Words per sentence: {analysis['words_per_sentence']:.1f}\n"
result += f"Complexity indicators found: {analysis['indicator_count']}\n"
if analysis['found_indicators']:
result += f"Indicators: {', '.join(analysis['found_indicators'])}\n"
result += f"Question count: {analysis['question_count']}\n"
result += f"\nEstimated complexity score: {analysis['estimated_complexity']}\n"
result += f"Recommended word limit per step: {analysis['estimated_complexity']}\n"
return result
async def main():
"""Main entry point for the MCP server."""
# Initialize example database
await example_db.get_examples("math") # This will trigger example loading if needed
# Print startup message
import sys
print("Chain of Draft MCP Server starting...", file=sys.stderr)
if __name__ == "__main__":
import sys
# Print debug information
print("Python version:", sys.version, file=sys.stderr)
print("Running in directory:", os.getcwd(), file=sys.stderr)
print("Environment variables:", dict(os.environ), file=sys.stderr)
try:
# Run the example initialization
asyncio.run(main())
# Start the server with stdio transport
print("Starting FastMCP server with stdio transport...", file=sys.stderr)
# Cannot use additional options directly, use environment variables instead
os.environ['MCP_DEBUG'] = '1'
os.environ['MCP_LOG_TRANSPORT'] = '1'
app.run(transport="stdio")
except Exception as e:
print(f"Error in server startup: {e}", file=sys.stderr)
import traceback
traceback.print_exc(file=sys.stderr)
sys.exit(1)
```
--------------------------------------------------------------------------------
/src/python/examples.py:
--------------------------------------------------------------------------------
```python
"""
Example database management for the Chain of Draft MCP server.
Stores and retrieves example problems and solutions.
"""
import os
from sqlalchemy import create_engine, Column, Integer, String, Text, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class Example(Base):
"""Database model for storing reasoning examples."""
__tablename__ = 'examples'
id = Column(Integer, primary_key=True)
problem = Column(Text, nullable=False)
reasoning = Column(Text, nullable=False)
answer = Column(String, nullable=False)
domain = Column(String, nullable=False)
approach = Column(String, nullable=False) # "CoD" or "CoT"
meta_data = Column(JSON, nullable=True) # Changed from metadata to avoid SQLAlchemy reserved keyword
class ExampleDatabase:
"""Manages a database of reasoning examples for few-shot prompting."""
def __init__(self, db_path=None):
"""Initialize the example database with a connection."""
if db_path is None:
# Default to SQLite in the current directory
db_path = os.environ.get("COD_EXAMPLES_DB", "cod_examples.db")
self.engine = create_engine(f"sqlite:///{db_path}")
Base.metadata.create_all(self.engine)
self.Session = sessionmaker(bind=self.engine)
# Initialize with some examples if empty
self._ensure_examples_exist()
def _ensure_examples_exist(self):
"""Check if examples exist and seed the database if empty."""
session = self.Session()
count = session.query(Example).count()
session.close()
if count == 0:
self._load_initial_examples()
def _load_initial_examples(self):
"""Load initial examples into the database."""
# Add some basic examples across domains
examples = [
# Math example (CoT)
{
"problem": "Jason had 20 lollipops. He gave Denny some lollipops. Now Jason has 12 lollipops. How many lollipops did Jason give to Denny?",
"reasoning": "Let's think through this step by step:\n1. Initially, Jason had 20 lollipops.\n2. After giving some to Denny, Jason now has 12 lollipops.\n3. To find out how many lollipops Jason gave to Denny, we need to calculate the difference between the initial number of lollipops and the remaining number.\n4. We can set up a simple subtraction problem: Initial number of lollipops - Remaining number of lollipops = Lollipops given to Denny\n5. Putting in the numbers: 20 - 12 = Lollipops given to Denny\n6. Solving the subtraction: 20 - 12 = 8",
"answer": "8 lollipops",
"domain": "math",
"approach": "CoT"
},
# Math example (CoD)
{
"problem": "Jason had 20 lollipops. He gave Denny some lollipops. Now Jason has 12 lollipops. How many lollipops did Jason give to Denny?",
"reasoning": "Initial: 20 lollipops\nRemaining: 12 lollipops\nGave away: 20-12=8 lollipops",
"answer": "8 lollipops",
"domain": "math",
"approach": "CoD"
},
# Logic example (CoT)
{
"problem": "A coin is heads up. John flips the coin. Mary flips the coin. Paul flips the coin. Susan does not flip the coin. Is the coin still heads up?",
"reasoning": "Let's track the state of the coin through each flip:\n1. Initially, the coin is heads up.\n2. John flips the coin, so it changes from heads to tails.\n3. Mary flips the coin, so it changes from tails to heads.\n4. Paul flips the coin, so it changes from heads to tails.\n5. Susan does not flip the coin, so it remains tails.\nTherefore, the coin is tails up, which means it is not still heads up.",
"answer": "No",
"domain": "logic",
"approach": "CoT"
},
# Logic example (CoD)
{
"problem": "A coin is heads up. John flips the coin. Mary flips the coin. Paul flips the coin. Susan does not flip the coin. Is the coin still heads up?",
"reasoning": "H→J flips→T\nT→M flips→H\nH→P flips→T\nT→S no flip→T\nFinal: tails",
"answer": "No",
"domain": "logic",
"approach": "CoD"
},
# Physics example (CoT)
{
"problem": "A car accelerates from 0 to 60 mph in 5 seconds. What is its acceleration in mph/s?",
"reasoning": "Let's solve this problem step by step:\n1. We know the initial velocity is 0 mph.\n2. The final velocity is 60 mph.\n3. The time taken is 5 seconds.\n4. Acceleration is the rate of change of velocity with respect to time.\n5. Using the formula: acceleration = (final velocity - initial velocity) / time\n6. Substituting the values: acceleration = (60 mph - 0 mph) / 5 seconds\n7. Simplifying: acceleration = 60 mph / 5 seconds = 12 mph/s",
"answer": "12 mph/s",
"domain": "physics",
"approach": "CoT"
},
# Physics example (CoD)
{
"problem": "A car accelerates from 0 to 60 mph in 5 seconds. What is its acceleration in mph/s?",
"reasoning": "a = Δv/Δt\na = (60-0)/5\na = 12 mph/s",
"answer": "12 mph/s",
"domain": "physics",
"approach": "CoD"
}
]
# Add examples to database
session = self.Session()
try:
for example in examples:
session.add(Example(**example))
session.commit()
finally:
session.close()
async def get_examples(self, domain, approach="CoD", limit=3):
"""Get examples for a specific domain and approach."""
session = self.Session()
try:
examples = session.query(Example).filter_by(
domain=domain, approach=approach
).limit(limit).all()
return [
{
"problem": ex.problem,
"reasoning": ex.reasoning,
"answer": ex.answer,
"domain": ex.domain,
"approach": ex.approach
}
for ex in examples
]
finally:
session.close()
async def add_example(self, problem, reasoning, answer, domain, approach="CoD", metadata=None):
"""Add a new example to the database."""
session = self.Session()
try:
example = Example(
problem=problem,
reasoning=reasoning,
answer=answer,
domain=domain,
approach=approach,
meta_data=metadata
)
session.add(example)
session.commit()
return example.id
finally:
session.close()
async def transform_cot_to_cod(self, cot_example, max_words_per_step=5):
"""Transform a CoT example into CoD format with word limit per step."""
# Extract steps from CoT
steps = self._extract_reasoning_steps(cot_example["reasoning"])
# Transform each step to be more concise
cod_steps = []
for step in steps:
# This could use an LLM to summarize, or rules-based approach
cod_step = self._summarize_step(step, max_words_per_step)
cod_steps.append(cod_step)
# Create CoD version
return {
"problem": cot_example["problem"],
"reasoning": "\n".join(cod_steps),
"answer": cot_example["answer"],
"domain": cot_example["domain"],
"approach": "CoD"
}
def _extract_reasoning_steps(self, reasoning):
"""Extract individual reasoning steps from CoT reasoning."""
# Simple approach: split by numbered steps or line breaks
if any(f"{i}." in reasoning for i in range(1, 10)):
# Numbered steps
import re
steps = re.split(r'\d+\.', reasoning)
return [s.strip() for s in steps if s.strip()]
else:
# Split by line
return [s.strip() for s in reasoning.split('\n') if s.strip()]
def _summarize_step(self, step, max_words):
"""Summarize a reasoning step to meet the word limit."""
# For now, just truncate - in production, would use an LLM to summarize
words = step.split()
if len(words) <= max_words:
return step
# Simple heuristic: keep first few words focusing on calculation/operation
return " ".join(words[:max_words])
async def get_example_count_by_domain(self):
"""Get count of examples by domain and approach."""
session = self.Session()
try:
from sqlalchemy import func
results = session.query(
Example.domain,
Example.approach,
func.count(Example.id).label("count")
).group_by(Example.domain, Example.approach).all()
return [
{
"domain": r[0],
"approach": r[1],
"count": r[2]
}
for r in results
]
finally:
session.close()
```
--------------------------------------------------------------------------------
/src/python/client.py:
--------------------------------------------------------------------------------
```python
"""
OpenAI-compatible client wrapper for the Chain of Draft MCP server.
Provides a drop-in replacement for OpenAI and Anthropic clients.
"""
import os
import time
import uuid
import anthropic
import openai
from mistralai.client import MistralClient
from mistralai.models.chat_completion import ChatMessage
import requests
from dotenv import load_dotenv
from analytics import AnalyticsService
from complexity import ComplexityEstimator
from examples import ExampleDatabase
from format import FormatEnforcer
from reasoning import ReasoningSelector, create_cod_prompt, create_cot_prompt
# Load environment variables
load_dotenv()
class UnifiedLLMClient:
"""
Unified client that supports multiple LLM providers (Anthropic, OpenAI, Mistral, Ollama).
"""
def __init__(self):
"""Initialize the appropriate client based on environment variables."""
self.provider = os.getenv("LLM_PROVIDER", "anthropic").lower()
self.model = os.getenv("LLM_MODEL", "claude-3-7-sonnet-latest")
# Initialize the appropriate client
if self.provider == "anthropic":
if not os.getenv("ANTHROPIC_API_KEY"):
raise ValueError("ANTHROPIC_API_KEY is required for Anthropic provider")
self.client = anthropic.Anthropic(
api_key=os.getenv("ANTHROPIC_API_KEY"),
base_url=os.getenv("ANTHROPIC_BASE_URL")
)
elif self.provider == "openai":
if not os.getenv("OPENAI_API_KEY"):
raise ValueError("OPENAI_API_KEY is required for OpenAI provider")
self.client = openai.OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL")
)
elif self.provider == "mistral":
if not os.getenv("MISTRAL_API_KEY"):
raise ValueError("MISTRAL_API_KEY is required for Mistral provider")
self.client = MistralClient(
api_key=os.getenv("MISTRAL_API_KEY")
)
elif self.provider == "ollama":
self.base_url = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434")
else:
raise ValueError(f"Unsupported LLM provider: {self.provider}")
async def get_available_models(self):
"""Get list of available models from the current provider."""
try:
if self.provider == "anthropic":
response = await self.client.models.list()
return [model.id for model in response.data]
elif self.provider == "openai":
response = await self.client.models.list()
return [model.id for model in response.data]
elif self.provider == "mistral":
response = await self.client.list_models()
return [model.id for model in response.data] if response.data else []
elif self.provider == "ollama":
response = requests.get(f"{self.base_url}/api/tags")
return [model["name"] for model in response.json()["models"]]
return []
except Exception as e:
print(f"Error fetching models: {e}")
return []
async def chat(self, messages, model=None, max_tokens=None, temperature=None):
"""
Send chat messages to the current provider.
Args:
messages: List of message dicts with 'role' and 'content'
model: Optional model override
max_tokens: Optional max tokens limit
temperature: Optional temperature setting
Returns:
Dict with response content and token usage
"""
model = model or self.model
try:
if self.provider == "anthropic":
response = await self.client.messages.create(
model=model,
messages=[{
"role": msg["role"],
"content": msg["content"]
} for msg in messages],
max_tokens=max_tokens,
temperature=temperature
)
return {
"content": response.content[0].text,
"usage": {
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens
}
}
elif self.provider == "openai":
response = await self.client.chat.completions.create(
model=model,
messages=messages,
max_tokens=max_tokens,
temperature=temperature
)
return {
"content": response.choices[0].message.content,
"usage": {
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens
}
}
elif self.provider == "mistral":
response = await self.client.chat(
model=model,
messages=[ChatMessage(
role=msg["role"],
content=msg["content"]
) for msg in messages],
max_tokens=max_tokens,
temperature=temperature
)
return {
"content": response.choices[0].message.content,
"usage": {
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens
}
}
elif self.provider == "ollama":
response = requests.post(
f"{self.base_url}/api/chat",
json={
"model": model,
"messages": messages,
"options": {
"num_predict": max_tokens,
"temperature": temperature
}
}
)
data = response.json()
return {
"content": data["message"]["content"],
"usage": {
"input_tokens": 0, # Ollama doesn't provide token counts
"output_tokens": 0
}
}
raise ValueError(f"Unsupported provider: {self.provider}")
except Exception as e:
print(f"Error in chat: {e}")
raise
class ChainOfDraftClient:
"""
Drop-in replacement for OpenAI client that uses Chain of Draft reasoning.
Provides both OpenAI and Anthropic-compatible interfaces.
"""
def __init__(self, api_key=None, base_url=None, **kwargs):
"""Initialize the client with optional API key and settings."""
# Initialize the unified LLM client
self.llm_client = UnifiedLLMClient()
# Initialize services
self.analytics = AnalyticsService()
self.complexity_estimator = ComplexityEstimator()
self.example_db = ExampleDatabase()
self.format_enforcer = FormatEnforcer()
self.reasoning_selector = ReasoningSelector(self.analytics)
# Default settings
self.default_settings = {
"max_words_per_step": 8,
"enforce_format": True,
"adaptive_word_limit": True,
"track_analytics": True,
"max_tokens": 200000
}
# Update with any provided kwargs
self.settings = {**self.default_settings, **kwargs}
# OpenAI-style completions
async def completions(self, model=None, prompt=None, **kwargs):
"""
OpenAI-compatible completions interface.
Args:
model: Model to use (default from settings)
prompt: The problem to solve
**kwargs: Additional parameters including domain
Returns:
OpenAI-style completion response
"""
if not prompt:
raise ValueError("Prompt is required")
# Extract reasoning problem from prompt
problem = prompt
# Determine domain from kwargs or infer
domain = kwargs.get("domain", "general")
# Process reasoning request
result = await self.solve_with_reasoning(
problem,
domain,
model=model or self.settings["model"],
**kwargs
)
# Format in OpenAI style response
return {
"id": f"cod-{uuid.uuid4()}",
"object": "completion",
"created": int(time.time()),
"model": model or self.settings["model"],
"choices": [{
"text": result["final_answer"],
"index": 0,
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": len(prompt.split()),
"completion_tokens": result["token_count"],
"total_tokens": len(prompt.split()) + result["token_count"]
},
# Add custom fields for CoD-specific data
"reasoning": result["reasoning_steps"],
"approach": result["approach"]
}
# ChatCompletions-style method
async def chat(self, model=None, messages=None, **kwargs):
"""
OpenAI-compatible chat completions interface.
Args:
model: Model to use (default from settings)
messages: Chat history with the last user message as the problem
**kwargs: Additional parameters including domain
Returns:
OpenAI-style chat completion response
"""
if not messages:
raise ValueError("Messages are required")
# Extract last user message as the problem
last_user_msg = next((m["content"] for m in reversed(messages)
if m["role"] == "user"), "")
if not last_user_msg:
raise ValueError("No user message found in the provided messages")
# Determine domain from kwargs or infer
domain = kwargs.get("domain", "general")
# Process reasoning request
result = await self.solve_with_reasoning(
last_user_msg,
domain,
model=model or self.settings["model"],
**kwargs
)
# Format in OpenAI style response
return {
"id": f"cod-{uuid.uuid4()}",
"object": "chat.completion",
"created": int(time.time()),
"model": model or self.settings["model"],
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": f"{result['reasoning_steps']}\n\n####\n{result['final_answer']}"
},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": sum(len(m.get("content", "").split()) for m in messages),
"completion_tokens": result["token_count"],
"total_tokens": sum(len(m.get("content", "").split()) for m in messages) + result["token_count"]
}
}
# Anthropic-style messages
async def messages(self, model=None, messages=None, **kwargs):
"""
Anthropic-compatible messages interface.
Args:
model: Model to use (default from settings)
messages: Chat history with the last user message as the problem
**kwargs: Additional parameters including domain
Returns:
Anthropic-style message response
"""
if not messages:
raise ValueError("Messages are required")
# Extract last user message as the problem
last_user_msg = next((m["content"] for m in reversed(messages)
if m["role"] == "user"), "")
if not last_user_msg:
raise ValueError("No user message found in the provided messages")
# Determine domain from kwargs or infer
domain = kwargs.get("domain", "general")
# Process reasoning request
result = await self.solve_with_reasoning(
last_user_msg,
domain,
model=model or self.settings["model"],
**kwargs
)
# Format in Anthropic style response
return {
"id": f"msg_{uuid.uuid4()}",
"type": "message",
"role": "assistant",
"model": model or self.settings["model"],
"content": [
{
"type": "text",
"text": f"{result['reasoning_steps']}\n\n####\n{result['final_answer']}"
}
],
"usage": {
"input_tokens": sum(len(m.get("content", "").split()) for m in messages),
"output_tokens": result["token_count"]
},
# Add custom fields
"reasoning_approach": result["approach"],
"word_limit": result["word_limit"]
}
# Core reasoning implementation
async def solve_with_reasoning(self, problem, domain="general", **kwargs):
"""
Solve a problem using the appropriate reasoning approach.
Args:
problem: The problem text
domain: Problem domain (math, code, logic, etc.)
**kwargs: Additional parameters and settings
Returns:
Dictionary with reasoning steps and answer
"""
start_time = time.time()
# Override settings with kwargs
local_settings = {**self.settings, **kwargs}
# Determine complexity and select approach
complexity = await self.complexity_estimator.estimate_complexity(problem, domain)
if local_settings.get("approach"):
# Manually specified approach
approach = local_settings["approach"]
approach_reason = "Manually specified"
else:
# Auto-select based on problem
approach, approach_reason = await self.reasoning_selector.select_approach(
problem, domain, complexity
)
# Determine word limit
if local_settings["adaptive_word_limit"] and approach == "CoD":
word_limit = complexity # Use estimated complexity as word limit
else:
word_limit = local_settings["max_words_per_step"]
# Get examples
examples = await self.example_db.get_examples(domain, approach)
# Create prompt based on approach
if approach == "CoD":
prompt = create_cod_prompt(problem, domain, word_limit, examples)
else:
prompt = create_cot_prompt(problem, domain, examples)
# Generate response from LLM
response = await self.llm_client.chat(
[{"role": "user", "content": prompt["user"]}],
model=local_settings.get("model", "claude-3-7-sonnet-latest"),
max_tokens=local_settings.get("max_tokens", 500),
temperature=local_settings.get("temperature", 0.7)
)
# Extract reasoning and answer
full_response = response["content"]
parts = full_response.split("####")
reasoning = parts[0].strip()
answer = parts[1].strip() if len(parts) > 1 else "No clear answer found"
# Apply format enforcement if needed
if local_settings["enforce_format"] and approach == "CoD":
reasoning = self.format_enforcer.enforce_word_limit(reasoning, word_limit)
adherence = self.format_enforcer.analyze_adherence(reasoning, word_limit)
else:
adherence = None
# Record analytics
if local_settings["track_analytics"]:
execution_time = (time.time() - start_time) * 1000 # ms
await self.analytics.record_inference(
problem=problem,
domain=domain,
approach=approach,
word_limit=word_limit,
tokens_used=len(full_response.split()),
execution_time=execution_time,
reasoning=reasoning,
answer=answer,
metadata={
"complexity": complexity,
"approach_reason": approach_reason,
"adherence": adherence
}
)
return {
"reasoning_steps": reasoning,
"final_answer": answer,
"token_count": len(full_response.split()),
"approach": approach,
"complexity": complexity,
"word_limit": word_limit
}
# Utility methods
async def get_performance_stats(self, domain=None):
"""Get performance statistics for CoD vs CoT approaches."""
return await self.analytics.get_performance_by_domain(domain)
async def get_token_reduction_stats(self):
"""Get token reduction statistics for CoD vs CoT."""
return await self.analytics.get_token_reduction_stats()
def update_settings(self, **kwargs):
"""Update the client settings."""
self.settings.update(kwargs)
return self.settings
```
--------------------------------------------------------------------------------
/src/index.ts:
--------------------------------------------------------------------------------
```typescript
#!/usr/bin/env node
/**
* Chain of Draft (CoD) MCP Server in TypeScript
* Implements the MCP server for Chain of Draft reasoning
*/
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { CallToolRequestSchema, ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js";
import { Anthropic } from "@anthropic-ai/sdk";
import OpenAI from "openai";
import { Mistral } from "@mistralai/mistralai";
import { Ollama, type ListResponse, type ModelResponse } from "ollama";
import dotenv from "dotenv";
import {
AnalyticsRecord,
PerformanceStats,
TokenReductionStats,
ComplexityAnalysis,
ChainOfDraftParams,
ChainOfDraftResult,
AnalyticsDatabase,
ComplexityEstimator,
FormatEnforcer,
ReasoningSelector,
ToolArguments
} from './types';
import { logger } from './utils/logger.js';
// Common interfaces for response types
interface LLMMessage {
role: 'user' | 'assistant' | 'system';
content: string;
}
interface LLMResponse {
content: string;
usage?: {
input_tokens: number;
output_tokens: number;
};
}
// Type for model responses
interface ModelInfo {
id: string;
name?: string;
}
// Load environment variables
dotenv.config();
class UnifiedLLMClient {
private anthropicClient?: Anthropic;
private openaiClient?: OpenAI;
private mistralClient?: Mistral;
private ollamaClient?: Ollama;
private provider: string;
private model: string;
constructor() {
// Load from environment variables with defaults
this.provider = process.env.LLM_PROVIDER?.toLowerCase() || 'anthropic';
this.model = process.env.LLM_MODEL || 'claude-3-7-sonnet-latest';
logger.devLog('Initializing LLM client', {
provider: this.provider,
model: this.model,
env: process.env.NODE_ENV
});
// Initialize the appropriate client based on provider
this.initializeClient();
}
private initializeClient() {
logger.devLog('Initializing LLM provider:', this.provider);
switch (this.provider) {
case 'anthropic':
if (!process.env.ANTHROPIC_API_KEY) {
throw new Error('ANTHROPIC_API_KEY is required for Anthropic provider');
}
this.anthropicClient = new Anthropic({
apiKey: process.env.ANTHROPIC_API_KEY,
baseURL: process.env.ANTHROPIC_BASE_URL
});
break;
case 'openai':
if (!process.env.OPENAI_API_KEY) {
throw new Error('OPENAI_API_KEY is required for OpenAI provider');
}
this.openaiClient = new OpenAI({
apiKey: process.env.OPENAI_API_KEY,
baseURL: process.env.OPENAI_BASE_URL
});
break;
case 'mistral':
if (!process.env.MISTRAL_API_KEY) {
throw new Error('MISTRAL_API_KEY is required for Mistral provider');
}
this.mistralClient = new Mistral({
apiKey: process.env.MISTRAL_API_KEY
});
break;
case 'ollama':
this.ollamaClient = new Ollama({
host: process.env.OLLAMA_BASE_URL || 'http://localhost:11434'
});
break;
default:
throw new Error(`Unsupported LLM provider: ${this.provider}`);
}
}
async getAvailableModels(): Promise<string[]> {
logger.devLog('Fetching available models');
try {
switch (this.provider) {
case 'anthropic': {
const response = await fetch(`${process.env.ANTHROPIC_BASE_URL || 'https://api.anthropic.com'}/v1/models`, {
headers: {
'x-api-key': process.env.ANTHROPIC_API_KEY || '',
'anthropic-version': '2023-06-01'
}
});
const data = await response.json();
return data.models.map((model: ModelInfo) => model.id);
}
case 'openai':
const openaiModels = await this.openaiClient!.models.list();
return openaiModels.data.map((model: ModelInfo) => model.id);
case 'mistral': {
const response = await this.mistralClient!.models.list();
return (response.data || []).map(model => model.id);
}
case 'ollama':
const ollamaModels = await this.ollamaClient!.list();
return ollamaModels.models.map((model: ModelResponse) => model.name);
default:
return [];
}
} catch (error) {
logger.error('Error fetching models:', error);
logger.devLog('Detailed error:', {
name: error instanceof Error ? error.name : 'Unknown',
message: error instanceof Error ? error.message : String(error),
stack: error instanceof Error ? error.stack : undefined
});
return [];
}
}
async chat(messages: LLMMessage[], options: {
model?: string;
max_tokens?: number;
temperature?: number;
} = {}): Promise<LLMResponse> {
const model = options.model || this.model;
logger.devLog('Initiating chat:', {
model,
messageCount: messages.length,
options
});
const max_tokens = options.max_tokens || undefined;
const temperature = options.temperature || undefined;
try {
switch (this.provider) {
case 'anthropic': {
const response = await this.anthropicClient!.messages.create({
model,
messages: messages.map(m => ({
role: m.role as 'user' | 'assistant',
content: m.content
})),
max_tokens: max_tokens || 1000,
temperature: temperature || 0.7
});
// Handle different content block types safely
const content = response.content[0];
let textContent = '';
try {
if (content && typeof content === 'object') {
// Try to access text property safely
const text = 'text' in content ? content.text : null;
textContent = text ? String(text) : JSON.stringify(content);
} else {
textContent = String(content);
}
} catch (e) {
textContent = JSON.stringify(content);
}
return {
content: textContent,
usage: {
input_tokens: response.usage?.input_tokens || 0,
output_tokens: response.usage?.output_tokens || 0
}
};
}
case 'openai': {
const response = await this.openaiClient!.chat.completions.create({
model,
messages,
max_tokens,
temperature
});
return {
content: response.choices[0].message.content || '',
usage: {
input_tokens: response.usage?.prompt_tokens || 0,
output_tokens: response.usage?.completion_tokens || 0
}
};
}
case 'mistral': {
const response = await this.mistralClient!.chat.complete({
model,
messages: messages.map(m => ({
role: m.role as 'user' | 'assistant',
content: m.content
})),
maxTokens: max_tokens,
temperature: temperature
});
const messageContent = response.choices?.[0]?.message?.content;
let content: string;
if (Array.isArray(messageContent)) {
content = messageContent.map(chunk =>
typeof chunk === 'string' ? chunk : JSON.stringify(chunk)
).join('');
} else if (typeof messageContent === 'string') {
content = messageContent;
} else {
throw new Error('Invalid content format in Mistral response');
}
return {
content,
usage: {
input_tokens: response.usage?.promptTokens || 0,
output_tokens: response.usage?.completionTokens || 0
}
};
}
case 'ollama': {
const response = await this.ollamaClient!.chat({
model,
messages,
options: {
num_predict: max_tokens,
temperature
}
});
return {
content: response.message.content,
usage: {
input_tokens: 0,
output_tokens: 0
}
};
}
default:
throw new Error(`Unsupported LLM provider: ${this.provider}`);
}
} catch (error) {
logger.error('Error in chat:', error);
logger.devLog('Chat error details:', {
name: error instanceof Error ? error.name : 'Unknown',
message: error instanceof Error ? error.message : String(error),
stack: error instanceof Error ? error.stack : undefined,
provider: this.provider,
model
});
throw error;
}
}
}
// Initialize the unified LLM client
const llmClient = new UnifiedLLMClient();
// Initialize database connection for analytics
// This is a simplified in-memory version
const analyticsDb: AnalyticsDatabase = {
records: [],
addRecord(record: AnalyticsRecord): number {
this.records.push({
...record,
timestamp: new Date()
});
return this.records.length;
},
getPerformanceByDomain(domain?: string): PerformanceStats[] {
let filtered = domain
? this.records.filter(r => r.domain === domain)
: this.records;
// Group by domain and approach
const grouped: { [key: string]: {
domain: string;
approach: string;
total_tokens: number;
total_time: number;
count: number;
accuracy: number | null;
}} = {};
for (const record of filtered) {
const key = `${record.domain}-${record.approach}`;
if (!grouped[key]) {
grouped[key] = {
domain: record.domain,
approach: record.approach,
total_tokens: 0,
total_time: 0,
count: 0,
accuracy: null
};
}
grouped[key].total_tokens += record.tokens_used;
grouped[key].total_time += record.execution_time_ms;
grouped[key].count += 1;
}
// Calculate averages
return Object.values(grouped).map(g => ({
domain: g.domain,
approach: g.approach,
avg_tokens: g.total_tokens / g.count,
avg_time_ms: g.total_time / g.count,
accuracy: g.accuracy,
count: g.count
}));
},
getTokenReductionStats(): TokenReductionStats[] {
// Group by domain
const domains: { [key: string]: {
cod_tokens: number[];
cot_tokens: number[];
}} = {};
for (const record of this.records) {
if (!domains[record.domain]) {
domains[record.domain] = {
cod_tokens: [],
cot_tokens: []
};
}
if (record.approach === 'CoD') {
domains[record.domain].cod_tokens.push(record.tokens_used);
} else if (record.approach === 'CoT') {
domains[record.domain].cot_tokens.push(record.tokens_used);
}
}
// Calculate reduction stats
return Object.entries(domains).map(([domain, data]) => {
const cod_avg = data.cod_tokens.length
? data.cod_tokens.reduce((a, b) => a + b, 0) / data.cod_tokens.length
: 0;
const cot_avg = data.cot_tokens.length
? data.cot_tokens.reduce((a, b) => a + b, 0) / data.cot_tokens.length
: 0;
let reduction = 0;
if (cot_avg > 0 && cod_avg > 0) {
reduction = 100 * (1 - (cod_avg / cot_avg));
}
return {
domain,
cod_avg_tokens: cod_avg,
cot_avg_tokens: cot_avg,
reduction_percentage: reduction
};
});
}
};
// Complexity estimation logic
const complexityEstimator: ComplexityEstimator = {
domainBaseLimits: {
math: 6,
logic: 5,
common_sense: 4,
physics: 7,
chemistry: 6,
biology: 5,
code: 8,
puzzle: 5,
general: 5
},
complexityIndicators: {
math: ['integral', 'derivative', 'equation', 'theorem', 'calculus', 'polynomial', 'algorithm'],
logic: ['if-then', 'premise', 'fallacy', 'syllogism', 'deduction', 'induction'],
physics: ['velocity', 'acceleration', 'quantum', 'momentum', 'thermodynamics'],
code: ['function', 'algorithm', 'recursive', 'complexity', 'optimization', 'edge case']
},
analyzeProblem(problem: string, domain: string): ComplexityAnalysis {
const wordCount = problem.split(/\s+/).filter(Boolean).length;
const sentences = problem.split(/[.!?]+/).filter(Boolean);
const sentenceCount = sentences.length;
const wordsPerSentence = sentenceCount > 0 ? wordCount / sentenceCount : 0;
// Count indicators
const indicators = this.complexityIndicators[domain] || this.complexityIndicators.general || [];
const lowerProblem = problem.toLowerCase();
const foundIndicators = indicators.filter(i => lowerProblem.includes(i.toLowerCase()));
// Count questions
const questionCount = (problem.match(/\?/g) || []).length;
// Estimate complexity
let complexity = this.domainBaseLimits[domain] || this.domainBaseLimits.general;
// Adjust for length
if (wordCount > 100) complexity += 2;
else if (wordCount > 50) complexity += 1;
// Adjust for sentences
if (wordsPerSentence > 20) complexity += 1;
// Adjust for indicators
complexity += Math.min(3, foundIndicators.length);
// Adjust for questions
complexity += Math.min(2, questionCount);
return {
word_count: wordCount,
sentence_count: sentenceCount,
words_per_sentence: wordsPerSentence,
indicator_count: foundIndicators.length,
found_indicators: foundIndicators,
question_count: questionCount,
estimated_complexity: complexity
};
}
};
// Format enforcement
const formatEnforcer: FormatEnforcer = {
enforceWordLimit(reasoning: string, maxWordsPerStep: number | null): string {
if (!maxWordsPerStep) return reasoning;
// Split into steps
const stepPattern = /(\d+\.\s*|Step\s+\d+:|\n-\s+|\n\*\s+|•\s+|^\s*-\s+|^\s*\*\s+)/;
let steps = reasoning.split(stepPattern).filter(Boolean);
// Process steps
const processed: string[] = [];
for (let i = 0; i < steps.length; i++) {
const step = steps[i];
// Check if this is a step marker
if (step.match(stepPattern)) {
processed.push(step);
continue;
}
// Enforce word limit on this step
const words = step.split(/\s+/).filter(Boolean);
if (words.length <= maxWordsPerStep) {
processed.push(step);
} else {
// Truncate to word limit
processed.push(words.slice(0, maxWordsPerStep).join(' '));
}
}
return processed.join('');
},
analyzeAdherence(reasoning: string, maxWordsPerStep: number) {
// Split into steps
const stepPattern = /(\d+\.\s*|Step\s+\d+:|\n-\s+|\n\*\s+|•\s+|^\s*-\s+|^\s*\*\s+)/;
let steps = reasoning.split(stepPattern).filter(Boolean);
let totalWords = 0;
let stepCount = 0;
let overLimitSteps = 0;
for (let i = 0; i < steps.length; i++) {
const step = steps[i];
// Skip step markers
if (step.match(stepPattern)) continue;
stepCount++;
const wordCount = step.split(/\s+/).filter(Boolean).length;
totalWords += wordCount;
if (wordCount > maxWordsPerStep) {
overLimitSteps++;
}
}
return {
step_count: stepCount,
total_words: totalWords,
avg_words_per_step: stepCount > 0 ? totalWords / stepCount : 0,
over_limit_steps: overLimitSteps,
adherence_percentage: stepCount > 0 ? (1 - (overLimitSteps / stepCount)) * 100 : 100
};
}
};
// Reasoning approach selector
const reasoningSelector: ReasoningSelector = {
defaultPreferences: {
math: {
complexity_threshold: 7,
accuracy_threshold: 0.85
},
code: {
complexity_threshold: 8,
accuracy_threshold: 0.9
},
general: {
complexity_threshold: 6,
accuracy_threshold: 0.8
}
},
selectApproach(domain: string, complexity: number, performanceStats: PerformanceStats[]): string {
// Default preferences for domain
const prefs = this.defaultPreferences[domain] || this.defaultPreferences.general;
// By default, use CoD for simpler problems
if (complexity < prefs.complexity_threshold) {
return 'CoD';
}
// For complex problems, check past performance if available
if (performanceStats && performanceStats.length > 0) {
const domainStats = performanceStats.filter(s => s.domain === domain);
// Check if CoD has good enough accuracy
const codStats = domainStats.find(s => s.approach === 'CoD');
if (codStats && codStats.accuracy && codStats.accuracy >= prefs.accuracy_threshold) {
return 'CoD';
}
// Otherwise, use CoT for complex problems
return 'CoT';
}
// Use CoT for complex problems by default
return complexity >= prefs.complexity_threshold ? 'CoT' : 'CoD';
}
};
// Prompt creation
function createCodPrompt(problem: string, domain: string, examples: string[] = [], wordLimit: number = 5): string {
return `
You will solve this ${domain} problem using the Chain of Draft technique, which uses very concise reasoning steps.
For each step, use at most ${wordLimit} words. Be extremely concise but clear.
PROBLEM:
${problem}
To solve this, create short reasoning steps, with each step using at most ${wordLimit} words.
After your reasoning, state your final answer clearly.
`.trim();
}
function createCotPrompt(problem: string, domain: string, examples: string[] = []): string {
return `
You will solve this ${domain} problem using detailed step-by-step reasoning.
PROBLEM:
${problem}
Think through this problem step-by-step with clear reasoning.
After your reasoning, state your final answer clearly.
`.trim();
}
// Chain of Draft client implementation
const chainOfDraftClient = {
async solveWithReasoning(params: ChainOfDraftParams): Promise<ChainOfDraftResult> {
const {
problem,
domain = 'general',
max_words_per_step = null,
approach = null,
enforce_format = true,
adaptive_word_limit = true
} = params;
logger.devLog('Starting Chain of Draft reasoning', params);
const startTime = Date.now();
// Analyze problem complexity
const analysis = complexityEstimator.analyzeProblem(problem, domain);
logger.devLog('Problem complexity analysis', analysis);
const complexity = analysis.estimated_complexity;
// Determine word limit
let wordLimit = max_words_per_step;
if (!wordLimit && adaptive_word_limit) {
wordLimit = complexity;
} else if (!wordLimit) {
wordLimit = complexityEstimator.domainBaseLimits[domain] || 5;
}
// Determine approach (CoD or CoT)
const performanceStats = analyticsDb.getPerformanceByDomain(domain);
const selectedApproach = approach ||
reasoningSelector.selectApproach(domain, complexity, performanceStats);
// Create prompt based on approach
const prompt = selectedApproach === 'CoD'
? createCodPrompt(problem, domain, [], wordLimit)
: createCotPrompt(problem, domain, []);
// Call LLM using unified client
const response = await llmClient.chat([
{ role: 'user', content: prompt }
], {
max_tokens: 1000,
temperature: 0.7
});
// Extract reasoning and answer
const fullText = response.content;
// Extract final answer (assuming it comes after the reasoning, often starts with "Answer:" or similar)
let reasoningSteps = fullText;
let finalAnswer = '';
// Common patterns for final answer sections
const answerPatterns = [
/(?:Final Answer|Answer|Therefore):?\s*(.*?)$/is,
/(?:In conclusion|To conclude|Thus|Hence|So),\s*(.*?)$/is,
/(?:The answer is|The result is|The solution is)\s*(.*?)$/is
];
// Try to extract the final answer with each pattern
for (const pattern of answerPatterns) {
const match = fullText.match(pattern);
if (match && match[1]) {
finalAnswer = match[1].trim();
reasoningSteps = fullText.substring(0, fullText.indexOf(match[0])).trim();
break;
}
}
// If no pattern matched, just use the last sentence
if (!finalAnswer) {
const sentences = fullText.split(/[.!?]+\s+/);
if (sentences.length > 1) {
finalAnswer = sentences.pop()!.trim();
reasoningSteps = sentences.join('. ') + '.';
}
}
// Apply format enforcement if needed
if (enforce_format && selectedApproach === 'CoD') {
reasoningSteps = formatEnforcer.enforceWordLimit(reasoningSteps, wordLimit);
}
// Calculate execution time
const executionTime = Date.now() - startTime;
// Get token count from response
const tokenCount = (response.usage?.input_tokens || 0) + (response.usage?.output_tokens || 0);
// Record analytics
analyticsDb.addRecord({
problem_id: problem.substring(0, 20),
problem_text: problem,
domain,
approach: selectedApproach,
word_limit: wordLimit,
tokens_used: tokenCount,
execution_time_ms: executionTime,
reasoning_steps: reasoningSteps,
answer: finalAnswer
});
// Return result
return {
approach: selectedApproach,
reasoning_steps: reasoningSteps,
final_answer: finalAnswer,
token_count: tokenCount,
word_limit: wordLimit,
complexity: complexity,
execution_time_ms: executionTime
};
}
};
// Define the Chain of Draft tool
const CHAIN_OF_DRAFT_TOOL = {
name: "chain_of_draft_solve",
description: "Solve a reasoning problem using Chain of Draft approach",
inputSchema: {
type: "object",
properties: {
problem: {
type: "string",
description: "The problem to solve"
},
domain: {
type: "string",
description: "Domain for context (math, logic, code, common-sense, etc.)"
},
max_words_per_step: {
type: "number",
description: "Maximum words per reasoning step"
},
approach: {
type: "string",
description: "Force 'CoD' or 'CoT' approach"
},
enforce_format: {
type: "boolean",
description: "Whether to enforce the word limit"
},
adaptive_word_limit: {
type: "boolean",
description: "Adjust word limits based on complexity"
}
},
required: ["problem"]
}
};
const MATH_TOOL = {
name: "math_solve",
description: "Solve a math problem using Chain of Draft reasoning",
inputSchema: {
type: "object",
properties: {
problem: {
type: "string",
description: "The math problem to solve"
},
approach: {
type: "string",
description: "Force 'CoD' or 'CoT' approach"
},
max_words_per_step: {
type: "number",
description: "Maximum words per reasoning step"
}
},
required: ["problem"]
}
};
const CODE_TOOL = {
name: "code_solve",
description: "Solve a coding problem using Chain of Draft reasoning",
inputSchema: {
type: "object",
properties: {
problem: {
type: "string",
description: "The coding problem to solve"
},
approach: {
type: "string",
description: "Force 'CoD' or 'CoT' approach"
},
max_words_per_step: {
type: "number",
description: "Maximum words per reasoning step"
}
},
required: ["problem"]
}
};
const LOGIC_TOOL = {
name: "logic_solve",
description: "Solve a logic problem using Chain of Draft reasoning",
inputSchema: {
type: "object",
properties: {
problem: {
type: "string",
description: "The logic problem to solve"
},
approach: {
type: "string",
description: "Force 'CoD' or 'CoT' approach"
},
max_words_per_step: {
type: "number",
description: "Maximum words per reasoning step"
}
},
required: ["problem"]
}
};
const PERFORMANCE_TOOL = {
name: "get_performance_stats",
description: "Get performance statistics for CoD vs CoT approaches",
inputSchema: {
type: "object",
properties: {
domain: {
type: "string",
description: "Filter for specific domain"
}
}
}
};
const TOKEN_TOOL = {
name: "get_token_reduction",
description: "Get token reduction statistics for CoD vs CoT",
inputSchema: {
type: "object",
properties: {}
}
};
const COMPLEXITY_TOOL = {
name: "analyze_problem_complexity",
description: "Analyze the complexity of a problem",
inputSchema: {
type: "object",
properties: {
problem: {
type: "string",
description: "The problem to analyze"
},
domain: {
type: "string",
description: "Problem domain"
}
},
required: ["problem"]
}
};
// Initialize the MCP server
const server = new Server({
name: "mcp-chain-of-draft-prompt-tool",
version: "1.0.0"
}, {
capabilities: {
tools: {},
},
});
// Expose available tools
server.setRequestHandler(ListToolsRequestSchema, async () => ({
tools: [
CHAIN_OF_DRAFT_TOOL,
MATH_TOOL,
CODE_TOOL,
LOGIC_TOOL,
PERFORMANCE_TOOL,
TOKEN_TOOL,
COMPLEXITY_TOOL
],
}));
// Handle tool calls
server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: toolArgs } = request.params;
const args = toolArgs as unknown as ToolArguments;
try {
// Chain of Draft solve
if (name === "chain_of_draft_solve" && args?.problem) {
const params: ChainOfDraftParams = {
problem: args.problem,
domain: args.domain,
max_words_per_step: args.max_words_per_step ?? null,
approach: args.approach,
enforce_format: args.enforce_format,
adaptive_word_limit: args.adaptive_word_limit
};
const result = await chainOfDraftClient.solveWithReasoning(params);
const formattedResponse =
`Chain of ${result.approach} reasoning (${result.word_limit} word limit):\n\n` +
`${result.reasoning_steps}\n\n` +
`Final answer: ${result.final_answer}\n\n` +
`Stats: ${result.token_count} tokens, ${result.execution_time_ms.toFixed(0)}ms, ` +
`complexity score: ${result.complexity}`;
return {
content: [{
type: "text",
text: formattedResponse
}]
};
}
// Math solver
if (name === "math_solve" && args?.problem) {
const params: ChainOfDraftParams = {
problem: args.problem,
domain: "math",
max_words_per_step: args.max_words_per_step ?? null,
approach: args.approach
};
const result = await chainOfDraftClient.solveWithReasoning(params);
const formattedResponse =
`Chain of ${result.approach} reasoning (${result.word_limit} word limit):\n\n` +
`${result.reasoning_steps}\n\n` +
`Final answer: ${result.final_answer}\n\n` +
`Stats: ${result.token_count} tokens, ${result.execution_time_ms.toFixed(0)}ms, ` +
`complexity score: ${result.complexity}`;
return {
content: [{
type: "text",
text: formattedResponse
}]
};
}
// Code solver
if (name === "code_solve" && args?.problem) {
const params: ChainOfDraftParams = {
problem: args.problem,
domain: "code",
max_words_per_step: args.max_words_per_step ?? null,
approach: args.approach
};
const result = await chainOfDraftClient.solveWithReasoning(params);
const formattedResponse =
`Chain of ${result.approach} reasoning (${result.word_limit} word limit):\n\n` +
`${result.reasoning_steps}\n\n` +
`Final answer: ${result.final_answer}\n\n` +
`Stats: ${result.token_count} tokens, ${result.execution_time_ms.toFixed(0)}ms, ` +
`complexity score: ${result.complexity}`;
return {
content: [{
type: "text",
text: formattedResponse
}]
};
}
// Logic solver
if (name === "logic_solve" && args?.problem) {
const params: ChainOfDraftParams = {
problem: args.problem,
domain: "logic",
max_words_per_step: args.max_words_per_step ?? null,
approach: args.approach
};
const result = await chainOfDraftClient.solveWithReasoning(params);
const formattedResponse =
`Chain of ${result.approach} reasoning (${result.word_limit} word limit):\n\n` +
`${result.reasoning_steps}\n\n` +
`Final answer: ${result.final_answer}\n\n` +
`Stats: ${result.token_count} tokens, ${result.execution_time_ms.toFixed(0)}ms, ` +
`complexity score: ${result.complexity}`;
return {
content: [{
type: "text",
text: formattedResponse
}]
};
}
// Performance stats
if (name === "get_performance_stats") {
const stats = analyticsDb.getPerformanceByDomain(args?.domain);
let result = "Performance Comparison (CoD vs CoT):\n\n";
if (!stats || stats.length === 0) {
result = "No performance data available yet.";
} else {
for (const stat of stats) {
result += `Domain: ${stat.domain}\n`;
result += `Approach: ${stat.approach}\n`;
result += `Average tokens: ${stat.avg_tokens.toFixed(1)}\n`;
result += `Average time: ${stat.avg_time_ms.toFixed(1)}ms\n`;
if (stat.accuracy !== null) {
result += `Accuracy: ${(stat.accuracy * 100).toFixed(1)}%\n`;
}
result += `Sample size: ${stat.count}\n\n`;
}
}
return {
content: [{
type: "text",
text: result
}]
};
}
// Token reduction
if (name === "get_token_reduction") {
const stats = analyticsDb.getTokenReductionStats();
let result = "Token Reduction Analysis:\n\n";
if (!stats || stats.length === 0) {
result = "No reduction data available yet.";
} else {
for (const stat of stats) {
result += `Domain: ${stat.domain}\n`;
result += `CoD avg tokens: ${stat.cod_avg_tokens.toFixed(1)}\n`;
result += `CoT avg tokens: ${stat.cot_avg_tokens.toFixed(1)}\n`;
result += `Reduction: ${stat.reduction_percentage.toFixed(1)}%\n\n`;
}
}
return {
content: [{
type: "text",
text: result
}]
};
}
// Complexity analysis
if (name === "analyze_problem_complexity" && args?.problem) {
const analysis = complexityEstimator.analyzeProblem(
args.problem,
args.domain || "general"
);
let result = `Complexity Analysis for ${args.domain || "general"} problem:\n\n`;
result += `Word count: ${analysis.word_count}\n`;
result += `Sentence count: ${analysis.sentence_count}\n`;
result += `Words per sentence: ${analysis.words_per_sentence.toFixed(1)}\n`;
result += `Complexity indicators found: ${analysis.indicator_count}\n`;
if (analysis.found_indicators && analysis.found_indicators.length > 0) {
result += `Indicators: ${analysis.found_indicators.join(", ")}\n`;
}
result += `Question count: ${analysis.question_count}\n`;
result += `\nEstimated complexity score: ${analysis.estimated_complexity}\n`;
result += `Recommended word limit per step: ${analysis.estimated_complexity}\n`;
return {
content: [{
type: "text",
text: result
}]
};
}
// Handle unknown tool
return {
content: [{
type: "text",
text: `Unknown tool: ${name}`
}],
isError: true
};
} catch (error) {
logger.error('Error executing tool:', error);
return {
content: [{
type: "text",
text: `Error executing tool ${name}: ${error instanceof Error ? error.message : String(error)}`
}],
isError: true
};
}
});
// Start the server
async function startServer() {
const transport = new StdioServerTransport();
logger.info('Chain of Draft MCP Server starting...');
try {
// Connect to the transport
await server.connect(transport);
logger.success('Server connected to transport');
// Keep the process alive
process.stdin.resume();
// Handle shutdown signals
const signals = ['SIGINT', 'SIGTERM', 'SIGQUIT'];
signals.forEach(signal => {
process.on(signal, async () => {
logger.info(`Received ${signal}, shutting down gracefully...`);
try {
await server.close();
logger.success('Server disconnected successfully');
process.exit(0);
} catch (error) {
logger.error('Error during shutdown:', error);
process.exit(1);
}
});
});
} catch (error) {
logger.error('Error starting server:', error);
process.exit(1);
}
}
// Run the server
startServer().catch(error => {
logger.error('Fatal error:', error);
process.exit(1);
});
process.on('uncaughtException', (error) => {
logger.error('Fatal error:', error);
process.exit(1);
});
```