torbjorn/my-pal-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP - communication memory

Browse Source
This commit is contained in:
Fahad
2025-06-10 19:16:51 +04:00
parent bb8a101dbf
commit f5060367a0
23 changed files with 2111 additions and 716 deletions
Download Patch File Download Diff File Expand all files Collapse all files

446
README.md
View File

@@ -11,10 +11,9 @@ The ultimate development partner for Claude - a Model Context Protocol server th
## Quick Navigation
- **Getting Started**
- [Quickstart](#quickstart-5-minutes) - Get running in 5 minutes
- [Docker Setup](#docker-setup-recommended) - Recommended approach
- [Traditional Setup](#option-b-traditional-setup) - Python-based setup
- [Quickstart](#quickstart-5-minutes) - Get running in 5 minutes with Docker
- [Available Tools](#available-tools) - Overview of all tools
- [AI-to-AI Conversations](#ai-to-ai-conversation-threading) - Multi-turn conversations
- **Tools Reference**
- [`chat`](#1-chat---general-development-chat--collaborative-thinking) - Collaborative thinking
@@ -72,231 +71,119 @@ The final implementation resulted in a 26% improvement in JSON parsing performan
### Prerequisites
**Recommended: Docker Setup (Works on all platforms)**
- Docker Desktop installed ([Download here](https://www.docker.com/products/docker-desktop/))
- Git
- **Windows users**: WSL2 is required for Claude Code CLI
- **Benefits**: No Python setup, consistent environment, easy updates, works everywhere
**Alternative: Traditional Python Setup**
- **Python 3.10 or higher** (required by the `mcp` package)
- Git
- **Windows users**: Must use WSL2 with Python installed inside WSL
- **Note**: More setup complexity, potential environment issues
### 1. Get a Gemini API Key
Visit [Google AI Studio](https://makersuite.google.com/app/apikey) and generate an API key. For best results with Gemini 2.5 Pro, use a paid API key as the free tier has limited access to the latest models.
### 2. Clone and Set Up the Repository
### 2. Clone and Set Up
```bash
# Clone to your preferred location
git clone https://github.com/BeehiveInnovations/gemini-mcp-server.git
cd gemini-mcp-server
# One-command setup (includes Redis for AI conversations)
./setup-docker.sh
```
**We strongly recommend Docker for the most reliable experience across all platforms.**
**What this does:**
- **Builds Docker images** with all dependencies (including Redis for conversation threading)
- **Creates .env file** (automatically uses `$GEMINI_API_KEY` if set in environment)
- **Starts Redis service** for AI-to-AI conversation memory
- **Starts MCP server** ready to connect
- **Shows exact Claude Desktop configuration** to copy
- **Multi-turn AI conversations** - Gemini can ask follow-up questions that persist across requests
#### Docker Setup (Recommended for all platforms)
### 3. Add Your API Key
```bash
# 1. Generate the .env file with your current directory as workspace
# macOS/Linux:
./setup-docker-env.sh
# Edit .env to add your Gemini API key (if not already set in environment)
nano .env
# Windows (PowerShell):
.\setup-docker-env.ps1
# The file will contain:
# GEMINI_API_KEY=your-gemini-api-key-here
# REDIS_URL=redis://redis:6379/0 (automatically configured)
# WORKSPACE_ROOT=/workspace (automatically configured)
```
**Important:** The setup script will:
- Create a `.env` file with your API key (automatically uses `$GEMINI_API_KEY` if already in your environment)
- **Automatically build the Docker image for you** - no manual build needed!
- Clean up any existing containers/images before building fresh
- **Display the exact Claude Desktop configuration to copy** - save this output for the next step, or configure [Claude Code directly](#5-connect-to-claude-code)
- Show you where to paste the configuration
**To update the app:** Simply run the setup script again - it will rebuild everything automatically.
**Docker File Access:** Docker containers can only access files within mounted directories. The generated configuration mounts your home directory by default. To access files elsewhere, modify the `-v` parameter in the configuration.
```bash
# 2. Edit .env to add your Gemini API key (if not already set in environment)
# The .env file will contain:
# WORKSPACE_ROOT=/your/current/directory (automatically set)
# GEMINI_API_KEY=your-gemini-api-key-here (automatically set if $GEMINI_API_KEY exists)
# 3. Copy the configuration from step 1 into Claude Desktop
```
**That's it!** The setup script handles everything - building the Docker image, setting up the environment, and configuring your API key.
#### Traditional Python Setup (Alternative)
```bash
# Run the setup script to install dependencies
# macOS/Linux:
./setup.sh
# Note: Claude Code requires WSL on Windows. See WSL setup instructions below.
```
**Note the full path** - you'll need it in the next step:
- **macOS/Linux**: `/Users/YOUR_USERNAME/gemini-mcp-server`
- **Windows (WSL)**: `/home/YOUR_WSL_USERNAME/gemini-mcp-server`
**Important**: The setup script will:
- Create a Python virtual environment
- Install all required dependencies (mcp, google-genai, etc.)
- Verify your Python installation
- Provide next steps for configuration
If you encounter any issues during setup, see the [Troubleshooting](#troubleshooting) section.
### 3. Configure Claude Desktop
Add the server to your `claude_desktop_config.json`:
### 4. Configure Claude Desktop
**Find your config file:**
- **macOS**: `~/Library/Application Support/Claude/claude_desktop_config.json`
- **Windows**: `%APPDATA%\Claude\claude_desktop_config.json` (configure for WSL usage) (configure for WSL usage)
- **Windows (WSL required)**: Access from WSL using `/mnt/c/Users/USERNAME/AppData/Roaming/Claude/claude_desktop_config.json`
**Or use Claude Desktop UI (macOS):**
- Open Claude Desktop
- Go to **Settings** → **Developer** → **Edit Config**
Choose your configuration based on your setup method:
**Or use Claude Code CLI (Recommended):**
```bash
# Add the MCP server directly via Claude Code CLI
claude mcp add gemini docker exec -i gemini-mcp-server-gemini-mcp-1
#### Docker Configuration (Recommended for all platforms)
# List your MCP servers to verify
claude mcp list
**How it works:** Claude Desktop launches Docker, which runs the MCP server in a container. The communication happens through stdin/stdout, just like running a regular command.
# Remove if needed
claude mcp remove gemini
```
#### Docker Configuration (Copy from setup script output)
The setup script shows you the exact configuration. It looks like this:
**All Platforms (macOS/Linux/Windows WSL):**
```json
{
"mcpServers": {
"gemini": {
"command": "docker",
"args": [
"run",
"--rm",
"-i",
"--env-file", "/path/to/gemini-mcp-server/.env",
"-v", "/path/to/your/project:/workspace:ro",
"gemini-mcp-server:latest"
"exec",
"-i",
"gemini-mcp-server-gemini-mcp-1"
]
}
}
}
```
**Important for Docker setup:**
- Replace `/path/to/gemini-mcp-server/.env` with the full path to your .env file
- Docker containers can ONLY access files within the mounted directory (`-v` parameter)
- The examples below mount your home directory for broad file access
- To access files elsewhere, change the mount path (e.g., `-v /specific/project:/workspace:ro`)
- The container runs temporarily when Claude needs it (no persistent Docker containers)
- Communication happens via stdio - Docker's `-i` flag connects the container's stdin/stdout to Claude
**How it works:**
- **Docker Compose services** run continuously in the background
- **Redis** automatically handles conversation memory between requests
- **AI-to-AI conversations** persist across multiple exchanges
- **File access** through mounted workspace directory
- **Fast communication** via `docker exec` to running container
**Path Format Notes:**
- **Windows users**: Use forward slashes `/` in Docker paths (e.g., `C:/Users/john/project`)
- Docker on Windows automatically handles both forward slashes and backslashes
- The setup scripts generate the correct format for your platform
**That's it!** The Docker setup handles all dependencies, Redis configuration, and service management automatically.
**Example for macOS/Linux:**
```json
{
"mcpServers": {
"gemini": {
"command": "docker",
"args": [
"run",
"--rm",
"-i",
"--env-file", "/path/to/gemini-mcp-server/.env",
"-e", "WORKSPACE_ROOT=/Users/YOUR_USERNAME",
"-e", "MCP_PROJECT_ROOT=/workspace",
"-v", "/Users/YOUR_USERNAME:/workspace:ro",
"gemini-mcp-server:latest"
]
}
}
}
```
## AI-to-AI Conversation Threading
**Example for Windows (WSL with Docker):**
```json
{
"mcpServers": {
"gemini": {
"command": "wsl.exe",
"args": ["-e", "docker", "run", "--rm", "-i", "--env-file", "/home/YOUR_WSL_USERNAME/gemini-mcp-server/.env", "-v", "/home/YOUR_WSL_USERNAME:/workspace:ro", "gemini-mcp-server:latest"]
}
}
}
```
This server supports **two-way conversations** between Claude and Gemini, enabling natural multi-turn discussions:
> **Note**: Run `setup-docker-env.sh` (macOS/Linux/WSL) or `setup-docker-env.ps1` (Windows PowerShell) to generate this configuration automatically with your paths.
**How it works:**
- Gemini can ask follow-up questions that you can answer naturally
- Each conversation maintains context across multiple exchanges
- All tools support conversation threading (chat, debug, thinkdeep, etc.)
- Conversations are automatically managed with Redis for persistence
#### Traditional Python Configuration (Alternative)
**Example:**
1. You: "Use gemini to analyze this authentication code"
2. Gemini: "I see potential security issues. Would you like me to examine the password hashing implementation?"
3. You: "Yes, check the password security"
4. Gemini: "Here's my analysis of the password handling..." (with full context)
**macOS/Linux:**
```json
{
"mcpServers": {
"gemini": {
"command": "/Users/YOUR_USERNAME/gemini-mcp-server/run_gemini.sh",
"env": {
"GEMINI_API_KEY": "your-gemini-api-key-here"
}
}
}
}
```
**Features:**
- Up to 5 exchanges per conversation
- 1-hour conversation expiry
- Thread-safe with Redis persistence
- Works across all Gemini tools seamlessly
**Windows (Using WSL):**
```json
{
"mcpServers": {
"gemini": {
"command": "wsl.exe",
"args": ["/home/YOUR_WSL_USERNAME/gemini-mcp-server/run_gemini.sh"],
"env": {
"GEMINI_API_KEY": "your-gemini-api-key-here"
}
}
}
}
```
### 4. Restart Claude Desktop
### 5. Restart Claude Desktop
Completely quit and restart Claude Desktop for the changes to take effect.
### 5. Connect to Claude Code
#### If you have Claude Desktop installed:
```bash
claude mcp add-from-claude-desktop -s user
```
#### If you only have Claude Code (no desktop app):
**For Traditional Setup (macOS/Linux):**
```bash
claude mcp add gemini -s user -e GEMINI_API_KEY=your-gemini-api-key-here -- /path/to/gemini-mcp-server/run_gemini.sh
```
**For Traditional Setup (Windows WSL):**
```bash
claude mcp add gemini -s user -e GEMINI_API_KEY=your-gemini-api-key-here -- /home/YOUR_WSL_USERNAME/gemini-mcp-server/run_gemini.sh
```
**For Docker Setup:**
```bash
claude mcp add gemini -s user -- docker run --rm -i --env-file /path/to/gemini-mcp-server/.env -v /home:/workspace:ro gemini-mcp-server:latest
```
Replace `/path/to/gemini-mcp-server` with the actual path where you cloned the repository.
### 6. Start Using It!
Just ask Claude naturally:
@@ -321,94 +208,6 @@ Just ask Claude naturally:
**Pro Tip:** You can control the depth of Gemini's analysis with thinking modes to manage token costs. For quick tasks use "minimal" or "low" to save tokens, for complex problems use "high" or "max" when quality matters more than cost. [Learn more about thinking modes](#thinking-modes---managing-token-costs--quality)
## Docker Setup (Recommended)
The Docker setup provides a consistent, hassle-free experience across all platforms without worrying about Python versions or dependencies.
### Why Docker is Recommended
- **Zero Python Setup**: No need to install Python or manage virtual environments
- **Consistent Environment**: Same behavior across macOS, Linux, and Windows WSL
- **Easy Updates**: Just run the setup script again to rebuild with latest changes
- **Isolated Dependencies**: No conflicts with your system Python packages
- **Platform Reliability**: Eliminates Python version conflicts and dependency issues
- **Windows Compatibility**: Works seamlessly with Claude Code's WSL requirement
### Quick Setup Guide
The setup scripts do all the heavy lifting for you:
1. **Run the setup script for your platform:**
```bash
# macOS/Linux:
./setup-docker-env.sh
# Windows (PowerShell):
.\setup-docker-env.ps1
# Windows (Command Prompt):
setup-docker-env.bat
```
2. **The script automatically:**
- Creates a `.env` file with your workspace and API key (if `$GEMINI_API_KEY` is set)
- **Builds the Docker image for you** - no manual `docker build` needed!
- Cleans up any old containers/images before building
- Displays the exact Claude Desktop configuration to copy
- Shows you where to paste it
3. **Edit `.env` to add your Gemini API key** (only if not already in your environment)
4. **Copy the configuration into Claude Desktop**
That's it! No manual Docker commands needed. **To update:** Just run the setup script again.
### How It Works
- **Path Translation**: The server automatically translates file paths between your host and the container
- **Workspace Mounting**: Your project directory is mounted to `/workspace` inside the container
- **stdio Communication**: Docker's `-i` flag preserves the MCP communication channel
### Testing Your Setup
```bash
# Test that the server starts correctly
docker run --rm -i --env-file .env -v "$(pwd):/workspace:ro" gemini-mcp-server:latest
# You should see "INFO:__main__:Gemini API key found"
# Press Ctrl+C to exit
```
## Windows Setup Guide
**Important**: Claude Code CLI does not support native Windows. You must use WSL (Windows Subsystem for Linux).
### WSL Setup (Required for Windows)
To use Claude Code on Windows, you must use WSL:
1. **Prerequisites**
- WSL2 installed with a Linux distribution (e.g., Ubuntu)
- Python installed in your WSL environment
- Project cloned inside WSL (recommended: `~/gemini-mcp-server`)
2. **Set up in WSL**
```bash
# Inside WSL terminal
cd ~/gemini-mcp-server
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
chmod +x run_gemini.sh
```
3. **Configure Claude Desktop** using the WSL configuration shown above
**Important WSL Notes:**
- For best performance, clone the repository inside WSL (`~/`) rather than on Windows (`/mnt/c/`)
- Ensure `run_gemini.sh` has Unix line endings (LF, not CRLF)
- If you have multiple WSL distributions, specify which one: `wsl.exe -d Ubuntu-22.04`
**Tools Overview:**
1. [`chat`](#1-chat---general-development-chat--collaborative-thinking) - Collaborative thinking and development conversations
2. [`thinkdeep`](#2-thinkdeep---extended-reasoning-partner) - Extended reasoning and problem-solving
@@ -1044,29 +843,6 @@ By default, the server allows access to files within your home directory. This i
This creates a sandbox limiting file access to only that directory and its subdirectories.
## Installation
1. Clone the repository:
```bash
git clone https://github.com/BeehiveInnovations/gemini-mcp-server.git
cd gemini-mcp-server
```
2. Create virtual environment:
```bash
python3 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
```
3. Install dependencies:
```bash
pip install -r requirements.txt
```
4. Set your Gemini API key:
```bash
export GEMINI_API_KEY="your-api-key-here"
```
## How System Prompts Work
@@ -1138,63 +914,51 @@ The CI pipeline works without any secrets and will pass all tests using mocked r
## Troubleshooting
### Windows/WSL Issues
**Important**: Claude Code CLI only supports WSL on Windows, not native Windows.
**Error: `spawn ENOENT` or execution issues**
This error occurs when Claude Desktop can't properly execute the server. Since Claude Code requires WSL:
1. **Ensure WSL is properly installed**: WSL2 with a Linux distribution (Ubuntu recommended)
2. **Use WSL configuration**: Always use `wsl.exe` in your Claude Desktop configuration
3. **Install dependencies in WSL**: Python and all packages must be installed inside WSL, not Windows
4. **Use WSL paths**: File paths should be WSL paths (`/home/username/...`) not Windows paths (`C:\...`)
**Testing your setup:**
- Verify WSL is working: `wsl.exe --list --verbose`
- Check Python in WSL: `wsl.exe python3 --version`
- Test server in WSL: `wsl.exe /home/YOUR_WSL_USERNAME/gemini-mcp-server/run_gemini.sh`
### Common Issues
**"ModuleNotFoundError: No module named 'mcp'" or "No matching distribution found for mcp"**
- This means either:
1. Python dependencies aren't installed - run the setup script
2. Your Python version is too old - the `mcp` package requires Python 3.10+
- **Solution**:
- First check your Python version: `python3 --version` or `python --version`
- If below 3.10, upgrade Python from https://python.org
- Then run the setup script:
- macOS/Linux: `./setup.sh`
- Windows: `setup.bat`
- If you still see this error, manually activate the virtual environment and install:
```bash
# macOS/Linux:
source venv/bin/activate
pip install -r requirements.txt
# Windows:
venv\Scripts\activate.bat
pip install -r requirements.txt
```
**"Virtual environment not found" warning**
- This is just a warning that can be ignored if dependencies are installed system-wide
- To fix: Run the setup script to create the virtual environment
**"GEMINI_API_KEY environment variable is required"**
- Ensure you've added your API key to the Claude Desktop configuration
- The key should be in the `env` section of your MCP server config
### Docker Issues
**"Connection failed" in Claude Desktop**
- Verify the command path is correct and uses proper escaping (`\\` for Windows paths)
- Ensure the script has execute permissions (Linux/macOS: `chmod +x run_gemini.sh`)
- Check Claude Desktop logs for detailed error messages
- Ensure Docker services are running: `docker compose ps`
- Check if the container name is correct: `docker ps` to see actual container names
- Verify your .env file has the correct GEMINI_API_KEY
**Performance issues with WSL**
- Files on Windows drives (`/mnt/c/`) are slower to access from WSL
- For best performance, clone the repository inside WSL (`~/gemini-mcp-server`)
**"GEMINI_API_KEY environment variable is required"**
- Edit your .env file and add your API key
- Restart services: `docker compose restart`
**Container fails to start**
- Check logs: `docker compose logs gemini-mcp`
- Ensure Docker has enough resources (memory/disk space)
- Try rebuilding: `docker compose build --no-cache`
**"spawn ENOENT" or execution issues**
- Verify the container is running: `docker compose ps`
- Check that Docker Desktop is running
- On Windows: Ensure WSL2 is properly configured for Docker
**Testing your Docker setup:**
```bash
# Check if services are running
docker compose ps
# Test manual connection
docker exec -i gemini-mcp-server-gemini-mcp-1 echo "Connection test"
# View logs
docker compose logs -f
```
**Conversation threading not working?**
If you're not seeing follow-up questions from Gemini:
```bash
# Check if Redis is running
docker compose logs redis
# Test conversation memory system
docker exec -i gemini-mcp-server-gemini-mcp-1 python debug_conversation.py
# Check for threading errors in logs
docker compose logs gemini-mcp | grep "threading failed"
```
## License

7
config.py
View File

@@ -8,6 +8,8 @@ constants used throughout the application.
Configuration values can be overridden by environment variables where appropriate.
"""
import os
# Version and metadata
# These values are used in server responses and for tracking releases
__version__ = "2.11.1" # Semantic versioning: MAJOR.MINOR.PATCH
@@ -49,3 +51,8 @@ TEMPERATURE_CREATIVE = 0.7 # For architecture, deep thinking
# to 50K characters (roughly ~10-12K tokens). Larger prompts must be sent
# as files to bypass MCP's token constraints.
MCP_PROMPT_SIZE_LIMIT = 50_000 # 50K characters
# Threading configuration
# Simple Redis-based conversation threading for stateless MCP environment
# Set REDIS_URL environment variable to connect to your Redis instance
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")

38
docker-compose.yml Normal file
View File

@@ -0,0 +1,38 @@
services:
redis:
image: redis:7-alpine
restart: unless-stopped
ports:
- "6379:6379"
volumes:
- redis_data:/data
command: redis-server --save 60 1 --loglevel warning --maxmemory 512mb --maxmemory-policy allkeys-lru
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 30s
timeout: 3s
retries: 3
deploy:
resources:
limits:
memory: 1G
reservations:
memory: 256M
gemini-mcp:
build: .
restart: unless-stopped
depends_on:
redis:
condition: service_healthy
environment:
- GEMINI_API_KEY=${GEMINI_API_KEY}
- REDIS_URL=redis://redis:6379/0
- WORKSPACE_ROOT=${WORKSPACE_ROOT:-/workspace}
volumes:
- ${HOME:-/tmp}:/workspace:ro
stdin_open: true
tty: true
volumes:
redis_data:

5
prompts/tool_prompts.py
View File

@@ -65,6 +65,7 @@ Be direct and technical. Assume Claude and the user are experienced developers w
deep, nuanced analysis rather than basic explanations. Your goal is to be the perfect
development partner that extends Claude's capabilities across diverse technology stacks."""
CODEREVIEW_PROMPT = """You are an expert code reviewer with deep knowledge of software engineering best practices.
Your expertise spans security, performance, maintainability, and architectural patterns.
@@ -133,6 +134,7 @@ Also provide:
- Top 3 priority fixes
- Positive aspects worth preserving"""
DEBUG_ISSUE_PROMPT = """You are an expert debugger and problem solver. Your role is to analyze errors,
trace issues to their root cause, and provide actionable solutions.
@@ -194,6 +196,7 @@ Steps to take regardless of root cause (e.g., error handling, logging)
Minimal steps to prevent this specific issue from recurring, directly related to the root cause identified.
**Targeted recommendations:** Specific to the exact problem resolved, not general best practices"""
ANALYZE_PROMPT = """You are an expert software analyst helping developers understand and work with code.
Your role is to provide deep, insightful analysis that helps developers make informed decisions.
@@ -271,6 +274,7 @@ Focus on (adapt priority based on project type and technology):
Be thorough but concise. Prioritize the most important findings and always provide
concrete examples and suggestions for improvement tailored to the specific technology stack."""
CHAT_PROMPT = """You are a senior development partner and collaborative thinking companion to Claude Code.
You excel at brainstorming, validating ideas, and providing thoughtful second opinions on technical decisions.
@@ -307,6 +311,7 @@ insights that might be missed in solo analysis. Think step by step through compl
and don't hesitate to explore tangential but relevant considerations that remain within the
project's technological and architectural boundaries."""
PRECOMMIT_PROMPT = """You are an expert code change analyst specializing in pre-commit review of git diffs.
Your role is to act as a seasoned senior developer performing a final review before code is committed.

1
requirements.txt
View File

@@ -1,6 +1,7 @@
mcp>=1.0.0
google-genai>=1.19.0
pydantic>=2.0.0
redis>=5.0.0
# Development dependencies
pytest>=7.4.0

37
run_gemini.sh
View File

@@ -1,37 +0,0 @@
#!/bin/bash
# Get the directory where this script is located
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# Check if virtual environment exists
if [ ! -d "$SCRIPT_DIR/venv" ]; then
echo "Virtual environment not found. Running setup..." >&2
# Check if setup.sh exists and is executable
if [ -f "$SCRIPT_DIR/setup.sh" ]; then
if [ ! -x "$SCRIPT_DIR/setup.sh" ]; then
chmod +x "$SCRIPT_DIR/setup.sh"
fi
# Run setup script
"$SCRIPT_DIR/setup.sh" >&2
# Check if setup was successful
if [ $? -ne 0 ]; then
echo "Setup failed. Please run setup.sh manually to see the error." >&2
exit 1
fi
else
echo "Error: setup.sh not found. Please ensure you have the complete repository." >&2
exit 1
fi
fi
# Activate virtual environment
source "$SCRIPT_DIR/venv/bin/activate"
# Change to script directory to ensure proper working directory
cd "$SCRIPT_DIR"
# Run the server
exec python "$SCRIPT_DIR/server.py"

131
server.py
View File

@@ -47,8 +47,11 @@ from tools import (
)
# Configure logging for server operations
# Set to INFO level to capture important operational messages without being too verbose
logging.basicConfig(level=logging.INFO)
# Set to DEBUG level to capture detailed operational messages for troubleshooting
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Create the MCP server instance with a unique name identifier
@@ -140,6 +143,10 @@ async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextCon
appropriate handlers. It supports both AI-powered tools (from TOOLS registry)
and utility tools (implemented as static functions).
Thread Context Reconstruction:
If the request contains a continuation_id, this function reconstructs
the conversation history and injects it into the tool's context.
Args:
name: The name of the tool to execute
arguments: Dictionary of arguments to pass to the tool
@@ -148,6 +155,10 @@ async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextCon
List of TextContent objects containing the tool's response
"""
# Handle thread context reconstruction if continuation_id is present
if "continuation_id" in arguments and arguments["continuation_id"]:
arguments = await reconstruct_thread_context(arguments)
# Route to AI-powered tools that require Gemini API calls
if name in TOOLS:
tool = TOOLS[name]
@@ -162,6 +173,122 @@ async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextCon
return [TextContent(type="text", text=f"Unknown tool: {name}")]
def get_follow_up_instructions(current_turn_count: int, max_turns: int = None) -> str:
"""
Generate dynamic follow-up instructions based on conversation turn count.
Args:
current_turn_count: Current number of turns in the conversation
max_turns: Maximum allowed turns before conversation ends (defaults to MAX_CONVERSATION_TURNS)
Returns:
Follow-up instructions to append to the tool prompt
"""
if max_turns is None:
from utils.conversation_memory import MAX_CONVERSATION_TURNS
max_turns = MAX_CONVERSATION_TURNS
if current_turn_count >= max_turns - 1:
# We're at or approaching the turn limit - no more follow-ups
return """
IMPORTANT: This is approaching the final exchange in this conversation thread.
Do NOT include any follow-up questions in your response. Provide your complete
final analysis and recommendations."""
else:
# Normal follow-up instructions
remaining_turns = max_turns - current_turn_count - 1
return f"""
🤝 CONVERSATION THREADING: You can continue this discussion with Claude! ({remaining_turns} exchanges remaining)
If you'd like to ask a follow-up question, explore a specific aspect deeper, or need clarification,
add this JSON block at the very end of your response:
```json
{{
"follow_up_question": "Would you like me to [specific action you could take]?",
"suggested_params": {{"files": ["relevant/files"], "focus_on": "specific area"}},
"ui_hint": "What this follow-up would accomplish"
}}
```
💡 Good follow-up opportunities:
- "Would you like me to examine the error handling in more detail?"
- "Should I analyze the performance implications of this approach?"
- "Would it be helpful to review the security aspects of this implementation?"
- "Should I dive deeper into the architecture patterns used here?"
Only ask follow-ups when they would genuinely add value to the discussion."""
async def reconstruct_thread_context(arguments: dict[str, Any]) -> dict[str, Any]:
"""
Reconstruct conversation context for thread continuation.
This function loads the conversation history from Redis and integrates it
into the request arguments to provide full context to the tool.
Args:
arguments: Original request arguments containing continuation_id
Returns:
Modified arguments with conversation history injected
"""
from utils.conversation_memory import add_turn, build_conversation_history, get_thread
continuation_id = arguments["continuation_id"]
# Get thread context from Redis
context = get_thread(continuation_id)
if not context:
logger.warning(f"Thread not found: {continuation_id}")
# Return error asking Claude to restart conversation with full context
raise ValueError(
f"Conversation thread '{continuation_id}' was not found or has expired. "
f"This may happen if the conversation was created more than 1 hour ago or if there was an issue with Redis storage. "
f"Please restart the conversation by providing your full question/prompt without the continuation_id parameter. "
f"This will create a new conversation thread that can continue with follow-up exchanges."
)
# Add user's new input to the conversation
user_prompt = arguments.get("prompt", "")
if user_prompt:
# Capture files referenced in this turn
user_files = arguments.get("files", [])
success = add_turn(continuation_id, "user", user_prompt, files=user_files)
if not success:
logger.warning(f"Failed to add user turn to thread {continuation_id}")
# Build conversation history
conversation_history = build_conversation_history(context)
# Add dynamic follow-up instructions based on turn count
follow_up_instructions = get_follow_up_instructions(len(context.turns))
# Merge original context with new prompt and follow-up instructions
original_prompt = arguments.get("prompt", "")
if conversation_history:
enhanced_prompt = (
f"{conversation_history}\n\n=== NEW USER INPUT ===\n{original_prompt}\n\n{follow_up_instructions}"
)
else:
enhanced_prompt = f"{original_prompt}\n\n{follow_up_instructions}"
# Update arguments with enhanced context
enhanced_arguments = arguments.copy()
enhanced_arguments["prompt"] = enhanced_prompt
# Merge original context parameters (files, etc.) with new request
if context.initial_context:
for key, value in context.initial_context.items():
if key not in enhanced_arguments and key not in ["temperature", "thinking_mode", "model"]:
enhanced_arguments[key] = value
logger.info(f"Reconstructed context for thread {continuation_id} (turn {len(context.turns)})")
return enhanced_arguments
async def handle_get_version() -> list[TextContent]:
"""
Get comprehensive version and configuration information about the server.

131
setup-docker-env.ps1
View File

@@ -1,131 +0,0 @@
# PowerShell script to set up .env file for Docker usage on Windows
Write-Host "Setting up .env file for Gemini MCP Server Docker..."
# Get the current working directory (absolute path)
$CurrentDir = Get-Location
# Check if .env already exists
if (Test-Path .env) {
Write-Host "Warning: .env file already exists! Skipping creation." -ForegroundColor Yellow
Write-Host ""
} else {
# Check if GEMINI_API_KEY is already set in environment
if ($env:GEMINI_API_KEY) {
$ApiKeyValue = $env:GEMINI_API_KEY
Write-Host "Found existing GEMINI_API_KEY in environment" -ForegroundColor Green
} else {
$ApiKeyValue = "your-gemini-api-key-here"
}
# Create the .env file
@"
# Gemini MCP Server Docker Environment Configuration
# Generated on $(Get-Date)
# Your Gemini API key (get one from https://makersuite.google.com/app/apikey)
# IMPORTANT: Replace this with your actual API key
GEMINI_API_KEY=$ApiKeyValue
"@ | Out-File -FilePath .env -Encoding utf8
Write-Host "Created .env file" -ForegroundColor Green
Write-Host ""
}
# Check if Docker is installed and running
if (!(Get-Command docker -ErrorAction SilentlyContinue)) {
Write-Host "⚠️ Docker is not installed. Please install Docker first." -ForegroundColor Yellow
Write-Host " Visit: https://docs.docker.com/get-docker/"
} else {
# Check if Docker daemon is running
try {
docker info 2>&1 | Out-Null
# Clean up and build Docker image
Write-Host ""
Write-Host "🐳 Building Docker image..." -ForegroundColor Blue
# Stop running containers
$runningContainers = docker ps -q --filter ancestor=gemini-mcp-server 2>$null
if ($runningContainers) {
Write-Host " - Stopping running containers..."
docker stop $runningContainers | Out-Null
}
# Remove containers
$allContainers = docker ps -aq --filter ancestor=gemini-mcp-server 2>$null
if ($allContainers) {
Write-Host " - Removing old containers..."
docker rm $allContainers | Out-Null
}
# Remove existing image
if (docker images | Select-String "gemini-mcp-server") {
Write-Host " - Removing old image..."
docker rmi gemini-mcp-server:latest 2>&1 | Out-Null
}
# Build fresh image
Write-Host " - Building fresh image with --no-cache..."
$result = docker build -t gemini-mcp-server:latest . --no-cache 2>&1
if ($LASTEXITCODE -eq 0) {
Write-Host "✅ Docker image built successfully!" -ForegroundColor Green
} else {
Write-Host "❌ Failed to build Docker image. Run 'docker build -t gemini-mcp-server:latest .' manually to see errors." -ForegroundColor Red
}
Write-Host ""
} catch {
Write-Host "⚠️ Docker daemon is not running. Please start Docker." -ForegroundColor Yellow
}
}
Write-Host "Next steps:"
if ($ApiKeyValue -eq "your-gemini-api-key-here") {
Write-Host "1. Edit .env and replace 'your-gemini-api-key-here' with your actual Gemini API key"
Write-Host "2. Copy this configuration to your Claude Desktop config:"
} else {
Write-Host "1. Copy this configuration to your Claude Desktop config:"
}
Write-Host ""
Write-Host "===== COPY BELOW THIS LINE =====" -ForegroundColor Cyan
Write-Host @"
{
"mcpServers": {
"gemini": {
"command": "$CurrentDir\gemini-mcp-docker.ps1"
}
}
}
"@
Write-Host "===== COPY ABOVE THIS LINE =====" -ForegroundColor Cyan
Write-Host ""
Write-Host "Alternative: If you prefer the direct Docker command:"
Write-Host @"
{
"mcpServers": {
"gemini": {
"command": "docker",
"args": [
"run",
"--rm",
"-i",
"--env-file", "$CurrentDir\.env",
"-e", "WORKSPACE_ROOT=$env:USERPROFILE",
"-v", "${env:USERPROFILE}:/workspace:ro",
"gemini-mcp-server:latest"
]
}
}
}
"@
Write-Host ""
Write-Host "Config file location:"
Write-Host " Windows: %APPDATA%\Claude\claude_desktop_config.json"
Write-Host ""
Write-Host "Note: This configuration mounts your user directory ($env:USERPROFILE)."
Write-Host "Docker can access any file within your user directory."
Write-Host ""
Write-Host "If you want to restrict access to a specific directory:"
Write-Host "Change both the mount (-v) and WORKSPACE_ROOT to match:"
Write-Host "Example: -v `"$CurrentDir:/workspace:ro`" and WORKSPACE_ROOT=$CurrentDir"
Write-Host "The container will automatically use /workspace as the sandbox boundary."

116
setup-docker-env.sh
View File

@@ -1,116 +0,0 @@
#!/bin/bash
# Helper script to set up .env file for Docker usage
echo "Setting up .env file for Gemini MCP Server Docker..."
# Get the current working directory (absolute path)
CURRENT_DIR=$(pwd)
# Check if .env already exists
if [ -f .env ]; then
echo "⚠️ .env file already exists! Skipping creation."
echo ""
else
# Check if GEMINI_API_KEY is already set in environment
if [ -n "$GEMINI_API_KEY" ]; then
API_KEY_VALUE="$GEMINI_API_KEY"
echo "✅ Found existing GEMINI_API_KEY in environment"
else
API_KEY_VALUE="your-gemini-api-key-here"
fi
# Create the .env file
cat > .env << EOF
# Gemini MCP Server Docker Environment Configuration
# Generated on $(date)
# Your Gemini API key (get one from https://makersuite.google.com/app/apikey)
# IMPORTANT: Replace this with your actual API key
GEMINI_API_KEY=$API_KEY_VALUE
EOF
echo "✅ Created .env file"
echo ""
fi
# Check if Docker is installed and running
if ! command -v docker &> /dev/null; then
echo "⚠️ Docker is not installed. Please install Docker first."
echo " Visit: https://docs.docker.com/get-docker/"
else
# Check if Docker daemon is running
if ! docker info &> /dev/null; then
echo "⚠️ Docker daemon is not running. Please start Docker."
else
# Clean up and build Docker image
echo ""
echo "🐳 Building Docker image..."
# Stop running containers
RUNNING_CONTAINERS=$(docker ps -q --filter ancestor=gemini-mcp-server 2>/dev/null)
if [ ! -z "$RUNNING_CONTAINERS" ]; then
echo " - Stopping running containers..."
docker stop $RUNNING_CONTAINERS >/dev/null 2>&1
fi
# Remove containers
ALL_CONTAINERS=$(docker ps -aq --filter ancestor=gemini-mcp-server 2>/dev/null)
if [ ! -z "$ALL_CONTAINERS" ]; then
echo " - Removing old containers..."
docker rm $ALL_CONTAINERS >/dev/null 2>&1
fi
# Remove existing image
if docker images | grep -q "gemini-mcp-server"; then
echo " - Removing old image..."
docker rmi gemini-mcp-server:latest >/dev/null 2>&1
fi
# Build fresh image
echo " - Building fresh image with --no-cache..."
if docker build -t gemini-mcp-server:latest . --no-cache >/dev/null 2>&1; then
echo "✅ Docker image built successfully!"
else
echo "❌ Failed to build Docker image. Run 'docker build -t gemini-mcp-server:latest .' manually to see errors."
fi
echo ""
fi
fi
echo "Next steps:"
if [ "$API_KEY_VALUE" = "your-gemini-api-key-here" ]; then
echo "1. Edit .env and replace 'your-gemini-api-key-here' with your actual Gemini API key"
echo "2. Copy this configuration to your Claude Desktop config:"
else
echo "1. Copy this configuration to your Claude Desktop config:"
fi
echo ""
echo "===== COPY BELOW THIS LINE ====="
echo "{"
echo " \"mcpServers\": {"
echo " \"gemini\": {"
echo " \"command\": \"docker\","
echo " \"args\": ["
echo " \"run\","
echo " \"--rm\","
echo " \"-i\","
echo " \"--env-file\", \"$CURRENT_DIR/.env\","
echo " \"-e\", \"WORKSPACE_ROOT=$HOME\","
echo " \"-v\", \"$HOME:/workspace:ro\","
echo " \"gemini-mcp-server:latest\""
echo " ]"
echo " }"
echo " }"
echo "}"
echo "===== COPY ABOVE THIS LINE ====="
echo ""
echo "Config file location:"
echo " macOS: ~/Library/Application Support/Claude/claude_desktop_config.json"
echo " Windows: %APPDATA%\\Claude\\claude_desktop_config.json"
echo ""
echo "Note: This configuration mounts your home directory ($HOME)."
echo "Docker can access any file within your home directory."
echo ""
echo "If you want to restrict access to a specific directory:"
echo "Change both the mount (-v) and WORKSPACE_ROOT to match:"
echo "Example: -v \"$CURRENT_DIR:/workspace:ro\" and WORKSPACE_ROOT=$CURRENT_DIR"
echo "The container will automatically use /workspace as the sandbox boundary."

149
setup-docker.sh Executable file
View File

@@ -0,0 +1,149 @@
#!/bin/bash
# Modern Docker setup script for Gemini MCP Server with Redis
# This script sets up the complete Docker environment including Redis for conversation threading
echo "🚀 Setting up Gemini MCP Server with Docker Compose..."
echo ""
# Get the current working directory (absolute path)
CURRENT_DIR=$(pwd)
# Check if .env already exists
if [ -f .env ]; then
echo "⚠️ .env file already exists! Updating if needed..."
echo ""
else
# Check if GEMINI_API_KEY is already set in environment
if [ -n "$GEMINI_API_KEY" ]; then
API_KEY_VALUE="$GEMINI_API_KEY"
echo "✅ Found existing GEMINI_API_KEY in environment"
else
API_KEY_VALUE="your-gemini-api-key-here"
fi
# Create the .env file
cat > .env << EOF
# Gemini MCP Server Docker Environment Configuration
# Generated on $(date)
# Your Gemini API key (get one from https://makersuite.google.com/app/apikey)
# IMPORTANT: Replace this with your actual API key
GEMINI_API_KEY=$API_KEY_VALUE
# Redis configuration (automatically set for Docker Compose)
REDIS_URL=redis://redis:6379/0
# Workspace root (automatically set for Docker Compose)
WORKSPACE_ROOT=/workspace
EOF
echo "✅ Created .env file with Redis configuration"
echo ""
fi
# Check if Docker and Docker Compose are installed
if ! command -v docker &> /dev/null; then
echo "❌ Docker is not installed. Please install Docker first."
echo " Visit: https://docs.docker.com/get-docker/"
exit 1
fi
if ! command -v docker-compose &> /dev/null && ! docker compose version &> /dev/null; then
echo "❌ Docker Compose is not installed. Please install Docker Compose first."
echo " Visit: https://docs.docker.com/compose/install/"
exit 1
fi
# Check if Docker daemon is running
if ! docker info &> /dev/null; then
echo "❌ Docker daemon is not running. Please start Docker."
exit 1
fi
# Use modern docker compose syntax if available, fall back to docker-compose
COMPOSE_CMD="docker compose"
if ! docker compose version &> /dev/null; then
COMPOSE_CMD="docker-compose"
fi
echo "🛠️ Building and starting services..."
echo ""
# Stop and remove existing containers
echo " - Stopping existing containers..."
$COMPOSE_CMD down --remove-orphans >/dev/null 2>&1
# Build and start services
echo " - Building Gemini MCP Server image..."
if $COMPOSE_CMD build --no-cache >/dev/null 2>&1; then
echo "✅ Docker image built successfully!"
else
echo "❌ Failed to build Docker image. Run '$COMPOSE_CMD build' manually to see errors."
exit 1
fi
echo " - Starting Redis and MCP services..."
if $COMPOSE_CMD up -d >/dev/null 2>&1; then
echo "✅ Services started successfully!"
else
echo "❌ Failed to start services. Run '$COMPOSE_CMD up -d' manually to see errors."
exit 1
fi
# Wait for services to be healthy
echo " - Waiting for Redis to be ready..."
sleep 3
# Check service status
if $COMPOSE_CMD ps --format table | grep -q "Up"; then
echo "✅ All services are running!"
else
echo "⚠️ Some services may not be running. Check with: $COMPOSE_CMD ps"
fi
echo ""
echo "📋 Service Status:"
$COMPOSE_CMD ps --format table
echo ""
echo "🔄 Next steps:"
if grep -q "your-gemini-api-key-here" .env; then
echo "1. Edit .env and replace 'your-gemini-api-key-here' with your actual Gemini API key"
echo "2. Restart services: $COMPOSE_CMD restart"
echo "3. Copy the configuration below to your Claude Desktop config:"
else
echo "1. Copy the configuration below to your Claude Desktop config:"
fi
echo ""
echo "===== CLAUDE DESKTOP CONFIGURATION ====="
echo "{"
echo " \"mcpServers\": {"
echo " \"gemini\": {"
echo " \"command\": \"docker\","
echo " \"args\": ["
echo " \"exec\","
echo " \"-i\","
echo " \"gemini-mcp-server-gemini-mcp-1\""
echo " ]"
echo " }"
echo " }"
echo "}"
echo "==========================================="
echo ""
echo "📁 Config file locations:"
echo " macOS: ~/Library/Application Support/Claude/claude_desktop_config.json"
echo " Windows (WSL): /mnt/c/Users/USERNAME/AppData/Roaming/Claude/claude_desktop_config.json"
echo ""
echo "🔧 Useful commands:"
echo " Start services: $COMPOSE_CMD up -d"
echo " Stop services: $COMPOSE_CMD down"
echo " View logs: $COMPOSE_CMD logs -f"
echo " Restart services: $COMPOSE_CMD restart"
echo " Service status: $COMPOSE_CMD ps"
echo ""
echo "🗃️ Redis for conversation threading is automatically configured and running!"
echo " All AI-to-AI conversations will persist between requests."

83
setup.sh
View File

@@ -1,83 +0,0 @@
#!/bin/bash
# Gemini MCP Server Setup Script
# This script helps users set up the virtual environment and install dependencies
echo "🚀 Gemini MCP Server Setup"
echo "========================="
# Get the directory where this script is located
SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
cd "$SCRIPT_DIR"
# Check if Python is installed
if ! command -v python3 &> /dev/null; then
echo "❌ Error: Python 3 is not installed."
echo "Please install Python 3.10 or higher from https://python.org"
exit 1
fi
# Display Python version
PYTHON_VERSION=$(python3 --version)
echo "✓ Found $PYTHON_VERSION"
# Check Python version is at least 3.10
PYTHON_VERSION_MAJOR=$(python3 -c 'import sys; print(sys.version_info.major)')
PYTHON_VERSION_MINOR=$(python3 -c 'import sys; print(sys.version_info.minor)')
if [ "$PYTHON_VERSION_MAJOR" -lt 3 ] || ([ "$PYTHON_VERSION_MAJOR" -eq 3 ] && [ "$PYTHON_VERSION_MINOR" -lt 10 ]); then
echo "❌ Error: Python 3.10 or higher is required (you have Python $PYTHON_VERSION_MAJOR.$PYTHON_VERSION_MINOR)"
echo ""
echo "The 'mcp' package requires Python 3.10 or newer."
echo "Please upgrade Python from https://python.org"
echo ""
echo "On macOS with Homebrew:"
echo " brew install python@3.10"
echo ""
echo "On Ubuntu/Debian:"
echo " sudo apt update && sudo apt install python3.10 python3.10-venv"
exit 1
fi
# Check if venv exists
if [ -d "venv" ]; then
echo "✓ Virtual environment already exists"
else
echo "📦 Creating virtual environment..."
python3 -m venv venv
if [ $? -eq 0 ]; then
echo "✓ Virtual environment created"
else
echo "❌ Error: Failed to create virtual environment"
exit 1
fi
fi
# Activate virtual environment
echo "🔧 Activating virtual environment..."
source venv/bin/activate
# Upgrade pip
echo "📦 Upgrading pip..."
pip install --upgrade pip
# Install requirements
echo "📦 Installing dependencies..."
pip install -r requirements.txt
if [ $? -eq 0 ]; then
echo ""
echo "✅ Setup completed successfully!"
echo ""
echo "Next steps:"
echo "1. Get your Gemini API key from: https://makersuite.google.com/app/apikey"
echo "2. Configure Claude Desktop with your API key (see README.md)"
echo "3. Restart Claude Desktop"
echo ""
echo "Note: The virtual environment has been activated for this session."
echo "The run_gemini.sh script will automatically activate it when needed."
else
echo "❌ Error: Failed to install dependencies"
echo "Please check the error messages above and try again."
exit 1
fi

413
tests/test_claude_continuation.py Normal file
View File

@@ -0,0 +1,413 @@
"""
Test suite for Claude continuation opportunities
Tests the system that offers Claude the opportunity to continue conversations
when Gemini doesn't explicitly ask a follow-up question.
"""
import json
from unittest.mock import Mock, patch
import pytest
from pydantic import Field
from tools.base import BaseTool, ToolRequest
from tools.models import ContinuationOffer, ToolOutput
from utils.conversation_memory import MAX_CONVERSATION_TURNS
class ContinuationRequest(ToolRequest):
"""Test request model with prompt field"""
prompt: str = Field(..., description="The prompt to analyze")
files: list[str] = Field(default_factory=list, description="Optional files to analyze")
class ClaudeContinuationTool(BaseTool):
"""Test tool for continuation functionality"""
def get_name(self) -> str:
return "test_continuation"
def get_description(self) -> str:
return "Test tool for Claude continuation"
def get_input_schema(self) -> dict:
return {
"type": "object",
"properties": {
"prompt": {"type": "string"},
"continuation_id": {"type": "string", "required": False},
},
}
def get_system_prompt(self) -> str:
return "Test system prompt"
def get_request_model(self):
return ContinuationRequest
async def prepare_prompt(self, request) -> str:
return f"System: {self.get_system_prompt()}\nUser: {request.prompt}"
class TestClaudeContinuationOffers:
"""Test Claude continuation offer functionality"""
def setup_method(self):
self.tool = ClaudeContinuationTool()
@patch("utils.conversation_memory.get_redis_client")
def test_new_conversation_offers_continuation(self, mock_redis):
"""Test that new conversations offer Claude continuation opportunity"""
mock_client = Mock()
mock_redis.return_value = mock_client
# Test request without continuation_id (new conversation)
request = ContinuationRequest(prompt="Analyze this code")
# Check continuation opportunity
continuation_data = self.tool._check_continuation_opportunity(request)
assert continuation_data is not None
assert continuation_data["remaining_turns"] == MAX_CONVERSATION_TURNS - 1
assert continuation_data["tool_name"] == "test_continuation"
def test_existing_conversation_no_continuation_offer(self):
"""Test that existing threaded conversations don't offer continuation"""
# Test request with continuation_id (existing conversation)
request = ContinuationRequest(
prompt="Continue analysis", continuation_id="12345678-1234-1234-1234-123456789012"
)
# Check continuation opportunity
continuation_data = self.tool._check_continuation_opportunity(request)
assert continuation_data is None
@patch("utils.conversation_memory.get_redis_client")
def test_create_continuation_offer_response(self, mock_redis):
"""Test creating continuation offer response"""
mock_client = Mock()
mock_redis.return_value = mock_client
request = ContinuationRequest(prompt="Test prompt")
content = "This is the analysis result."
continuation_data = {"remaining_turns": 4, "tool_name": "test_continuation"}
# Create continuation offer response
response = self.tool._create_continuation_offer_response(content, continuation_data, request)
assert isinstance(response, ToolOutput)
assert response.status == "continuation_available"
assert response.content == content
assert response.continuation_offer is not None
offer = response.continuation_offer
assert isinstance(offer, ContinuationOffer)
assert offer.remaining_turns == 4
assert "continuation_id" in offer.suggested_tool_params
assert "You have 4 more exchange(s) available" in offer.message_to_user
@patch("utils.conversation_memory.get_redis_client")
async def test_full_response_flow_with_continuation_offer(self, mock_redis):
"""Test complete response flow that creates continuation offer"""
mock_client = Mock()
mock_redis.return_value = mock_client
# Mock the model to return a response without follow-up question
with patch.object(self.tool, "create_model") as mock_create_model:
mock_model = Mock()
mock_response = Mock()
mock_response.candidates = [
Mock(
content=Mock(parts=[Mock(text="Analysis complete. The code looks good.")]),
finish_reason="STOP",
)
]
mock_model.generate_content.return_value = mock_response
mock_create_model.return_value = mock_model
# Execute tool with new conversation
arguments = {"prompt": "Analyze this code"}
response = await self.tool.execute(arguments)
# Parse response
assert len(response) == 1
response_data = json.loads(response[0].text)
# Debug output
if response_data.get("status") == "error":
print(f"Error content: {response_data.get('content')}")
assert response_data["status"] == "continuation_available"
assert response_data["content"] == "Analysis complete. The code looks good."
assert "continuation_offer" in response_data
offer = response_data["continuation_offer"]
assert "continuation_id" in offer
assert offer["remaining_turns"] == MAX_CONVERSATION_TURNS - 1
assert "You have" in offer["message_to_user"]
assert "more exchange(s) available" in offer["message_to_user"]
@patch("utils.conversation_memory.get_redis_client")
async def test_gemini_follow_up_takes_precedence(self, mock_redis):
"""Test that Gemini follow-up questions take precedence over continuation offers"""
mock_client = Mock()
mock_redis.return_value = mock_client
# Mock the model to return a response WITH follow-up question
with patch.object(self.tool, "create_model") as mock_create_model:
mock_model = Mock()
mock_response = Mock()
mock_response.candidates = [
Mock(
content=Mock(
parts=[
Mock(
text="""Analysis complete. The code looks good.
```json
{
"follow_up_question": "Would you like me to examine the error handling patterns?",
"suggested_params": {"files": ["/src/error_handler.py"]},
"ui_hint": "Examining error handling would help ensure robustness"
}
```"""
)
]
),
finish_reason="STOP",
)
]
mock_model.generate_content.return_value = mock_response
mock_create_model.return_value = mock_model
# Execute tool
arguments = {"prompt": "Analyze this code"}
response = await self.tool.execute(arguments)
# Parse response
response_data = json.loads(response[0].text)
# Should be follow-up, not continuation offer
assert response_data["status"] == "requires_continuation"
assert "follow_up_request" in response_data
assert response_data.get("continuation_offer") is None
@patch("utils.conversation_memory.get_redis_client")
async def test_threaded_conversation_no_continuation_offer(self, mock_redis):
"""Test that threaded conversations don't get continuation offers"""
mock_client = Mock()
mock_redis.return_value = mock_client
# Mock existing thread context
from utils.conversation_memory import ThreadContext
thread_context = ThreadContext(
thread_id="12345678-1234-1234-1234-123456789012",
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:01:00Z",
tool_name="test_continuation",
turns=[],
initial_context={"prompt": "Previous analysis"},
)
mock_client.get.return_value = thread_context.model_dump_json()
# Mock the model
with patch.object(self.tool, "create_model") as mock_create_model:
mock_model = Mock()
mock_response = Mock()
mock_response.candidates = [
Mock(
content=Mock(parts=[Mock(text="Continued analysis complete.")]),
finish_reason="STOP",
)
]
mock_model.generate_content.return_value = mock_response
mock_create_model.return_value = mock_model
# Execute tool with continuation_id
arguments = {"prompt": "Continue the analysis", "continuation_id": "12345678-1234-1234-1234-123456789012"}
response = await self.tool.execute(arguments)
# Parse response
response_data = json.loads(response[0].text)
# Should be regular success, not continuation offer
assert response_data["status"] == "success"
assert response_data.get("continuation_offer") is None
def test_max_turns_reached_no_continuation_offer(self):
"""Test that no continuation is offered when max turns would be exceeded"""
# Mock MAX_CONVERSATION_TURNS to be 1 for this test
with patch("utils.conversation_memory.MAX_CONVERSATION_TURNS", 1):
request = ContinuationRequest(prompt="Test prompt")
# Check continuation opportunity
continuation_data = self.tool._check_continuation_opportunity(request)
# Should be None because remaining_turns would be 0
assert continuation_data is None
@patch("utils.conversation_memory.get_redis_client")
def test_continuation_offer_thread_creation_failure_fallback(self, mock_redis):
"""Test fallback to normal response when thread creation fails"""
# Mock Redis to fail
mock_client = Mock()
mock_client.setex.side_effect = Exception("Redis failure")
mock_redis.return_value = mock_client
request = ContinuationRequest(prompt="Test prompt")
content = "Analysis result"
continuation_data = {"remaining_turns": 4, "tool_name": "test_continuation"}
# Should fallback to normal response
response = self.tool._create_continuation_offer_response(content, continuation_data, request)
assert response.status == "success"
assert response.content == content
assert response.continuation_offer is None
@patch("utils.conversation_memory.get_redis_client")
def test_continuation_offer_message_format(self, mock_redis):
"""Test that continuation offer message is properly formatted for Claude"""
mock_client = Mock()
mock_redis.return_value = mock_client
request = ContinuationRequest(prompt="Analyze architecture")
content = "Architecture analysis complete."
continuation_data = {"remaining_turns": 3, "tool_name": "test_continuation"}
response = self.tool._create_continuation_offer_response(content, continuation_data, request)
offer = response.continuation_offer
message = offer.message_to_user
# Check message contains key information for Claude
assert "continue this analysis" in message
assert "continuation_id" in message
assert "test_continuation tool call" in message
assert "3 more exchange(s)" in message
# Check suggested params are properly formatted
suggested_params = offer.suggested_tool_params
assert "continuation_id" in suggested_params
assert "prompt" in suggested_params
assert isinstance(suggested_params["continuation_id"], str)
@patch("utils.conversation_memory.get_redis_client")
def test_continuation_offer_metadata(self, mock_redis):
"""Test that continuation offer includes proper metadata"""
mock_client = Mock()
mock_redis.return_value = mock_client
request = ContinuationRequest(prompt="Test")
content = "Test content"
continuation_data = {"remaining_turns": 2, "tool_name": "test_continuation"}
response = self.tool._create_continuation_offer_response(content, continuation_data, request)
metadata = response.metadata
assert metadata["tool_name"] == "test_continuation"
assert metadata["remaining_turns"] == 2
assert "thread_id" in metadata
assert len(metadata["thread_id"]) == 36 # UUID length
class TestContinuationIntegration:
"""Integration tests for continuation offers with conversation memory"""
def setup_method(self):
self.tool = ClaudeContinuationTool()
@patch("utils.conversation_memory.get_redis_client")
def test_continuation_offer_creates_proper_thread(self, mock_redis):
"""Test that continuation offers create properly formatted threads"""
mock_client = Mock()
mock_redis.return_value = mock_client
# Mock the get call that add_turn makes to retrieve the existing thread
# We'll set this up after the first setex call
def side_effect_get(key):
# Return the context from the first setex call
if mock_client.setex.call_count > 0:
first_call_data = mock_client.setex.call_args_list[0][0][2]
return first_call_data
return None
mock_client.get.side_effect = side_effect_get
request = ContinuationRequest(prompt="Initial analysis", files=["/test/file.py"])
content = "Analysis result"
continuation_data = {"remaining_turns": 4, "tool_name": "test_continuation"}
response = self.tool._create_continuation_offer_response(content, continuation_data, request)
# Verify thread creation was called (should be called twice: create_thread + add_turn)
assert mock_client.setex.call_count == 2
# Check the first call (create_thread)
first_call = mock_client.setex.call_args_list[0]
thread_key = first_call[0][0]
assert thread_key.startswith("thread:")
assert len(thread_key.split(":")[-1]) == 36 # UUID length
# Check the second call (add_turn) which should have the assistant response
second_call = mock_client.setex.call_args_list[1]
thread_data = second_call[0][2]
thread_context = json.loads(thread_data)
assert thread_context["tool_name"] == "test_continuation"
assert len(thread_context["turns"]) == 1 # Assistant's response added
assert thread_context["turns"][0]["role"] == "assistant"
assert thread_context["turns"][0]["content"] == content
assert thread_context["turns"][0]["files"] == ["/test/file.py"] # Files from request
assert thread_context["initial_context"]["prompt"] == "Initial analysis"
assert thread_context["initial_context"]["files"] == ["/test/file.py"]
@patch("utils.conversation_memory.get_redis_client")
def test_claude_can_use_continuation_id(self, mock_redis):
"""Test that Claude can use the provided continuation_id in subsequent calls"""
mock_client = Mock()
mock_redis.return_value = mock_client
# Step 1: Initial request creates continuation offer
request1 = ToolRequest(prompt="Analyze code structure")
continuation_data = {"remaining_turns": 4, "tool_name": "test_continuation"}
response1 = self.tool._create_continuation_offer_response(
"Structure analysis done.", continuation_data, request1
)
thread_id = response1.continuation_offer.continuation_id
# Step 2: Mock the thread context for Claude's follow-up
from utils.conversation_memory import ConversationTurn, ThreadContext
existing_context = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:01:00Z",
tool_name="test_continuation",
turns=[
ConversationTurn(
role="assistant",
content="Structure analysis done.",
timestamp="2023-01-01T00:00:30Z",
tool_name="test_continuation",
)
],
initial_context={"prompt": "Analyze code structure"},
)
mock_client.get.return_value = existing_context.model_dump_json()
# Step 3: Claude uses continuation_id
request2 = ToolRequest(prompt="Now analyze the performance aspects", continuation_id=thread_id)
# This should NOT offer another continuation (already threaded)
continuation_data2 = self.tool._check_continuation_opportunity(request2)
assert continuation_data2 is None
if __name__ == "__main__":
pytest.main([__file__])

721
tests/test_conversation_memory.py Normal file
View File

@@ -0,0 +1,721 @@
"""
Test suite for conversation memory system
Tests the Redis-based conversation persistence needed for AI-to-AI multi-turn
discussions in stateless MCP environments.
"""
from unittest.mock import Mock, patch
import pytest
from server import get_follow_up_instructions
from utils.conversation_memory import (
MAX_CONVERSATION_TURNS,
ConversationTurn,
ThreadContext,
add_turn,
build_conversation_history,
create_thread,
get_thread,
)
class TestConversationMemory:
"""Test the conversation memory system for stateless MCP requests"""
@patch("utils.conversation_memory.get_redis_client")
def test_create_thread(self, mock_redis):
"""Test creating a new thread"""
mock_client = Mock()
mock_redis.return_value = mock_client
thread_id = create_thread("chat", {"prompt": "Hello", "files": ["/test.py"]})
assert thread_id is not None
assert len(thread_id) == 36 # UUID4 length
# Verify Redis was called
mock_client.setex.assert_called_once()
call_args = mock_client.setex.call_args
assert call_args[0][0] == f"thread:{thread_id}" # key
assert call_args[0][1] == 3600 # TTL
@patch("utils.conversation_memory.get_redis_client")
def test_get_thread_valid(self, mock_redis):
"""Test retrieving an existing thread"""
mock_client = Mock()
mock_redis.return_value = mock_client
test_uuid = "12345678-1234-1234-1234-123456789012"
# Create valid ThreadContext and serialize it
context_obj = ThreadContext(
thread_id=test_uuid,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:01:00Z",
tool_name="chat",
turns=[],
initial_context={"prompt": "test"},
)
mock_client.get.return_value = context_obj.model_dump_json()
context = get_thread(test_uuid)
assert context is not None
assert context.thread_id == test_uuid
assert context.tool_name == "chat"
mock_client.get.assert_called_once_with(f"thread:{test_uuid}")
@patch("utils.conversation_memory.get_redis_client")
def test_get_thread_invalid_uuid(self, mock_redis):
"""Test handling invalid UUID"""
context = get_thread("invalid-uuid")
assert context is None
@patch("utils.conversation_memory.get_redis_client")
def test_get_thread_not_found(self, mock_redis):
"""Test handling thread not found"""
mock_client = Mock()
mock_redis.return_value = mock_client
mock_client.get.return_value = None
context = get_thread("12345678-1234-1234-1234-123456789012")
assert context is None
@patch("utils.conversation_memory.get_redis_client")
def test_add_turn_success(self, mock_redis):
"""Test adding a turn to existing thread"""
mock_client = Mock()
mock_redis.return_value = mock_client
test_uuid = "12345678-1234-1234-1234-123456789012"
# Create valid ThreadContext
context_obj = ThreadContext(
thread_id=test_uuid,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:01:00Z",
tool_name="chat",
turns=[],
initial_context={"prompt": "test"},
)
mock_client.get.return_value = context_obj.model_dump_json()
success = add_turn(test_uuid, "user", "Hello there")
assert success is True
# Verify Redis get and setex were called
mock_client.get.assert_called_once()
mock_client.setex.assert_called_once()
@patch("utils.conversation_memory.get_redis_client")
def test_add_turn_max_limit(self, mock_redis):
"""Test turn limit enforcement"""
mock_client = Mock()
mock_redis.return_value = mock_client
test_uuid = "12345678-1234-1234-1234-123456789012"
# Create thread with MAX_CONVERSATION_TURNS turns (at limit)
turns = [
ConversationTurn(role="user", content=f"Turn {i}", timestamp="2023-01-01T00:00:00Z")
for i in range(MAX_CONVERSATION_TURNS)
]
context_obj = ThreadContext(
thread_id=test_uuid,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:01:00Z",
tool_name="chat",
turns=turns,
initial_context={"prompt": "test"},
)
mock_client.get.return_value = context_obj.model_dump_json()
success = add_turn(test_uuid, "user", "This should fail")
assert success is False
def test_build_conversation_history(self):
"""Test building conversation history format with files and speaker identification"""
test_uuid = "12345678-1234-1234-1234-123456789012"
turns = [
ConversationTurn(
role="user",
content="What is Python?",
timestamp="2023-01-01T00:00:00Z",
files=["/home/user/main.py", "/home/user/docs/readme.md"],
),
ConversationTurn(
role="assistant",
content="Python is a programming language",
timestamp="2023-01-01T00:01:00Z",
follow_up_question="Would you like examples?",
files=["/home/user/examples/"],
tool_name="chat",
),
]
context = ThreadContext(
thread_id=test_uuid,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:01:00Z",
tool_name="chat",
turns=turns,
initial_context={},
)
history = build_conversation_history(context)
# Test basic structure
assert "CONVERSATION HISTORY" in history
assert f"Thread: {test_uuid}" in history
assert "Tool: chat" in history
assert f"Turn 2/{MAX_CONVERSATION_TURNS}" in history
# Test speaker identification
assert "--- Turn 1 (Claude) ---" in history
assert "--- Turn 2 (Gemini using chat) ---" in history
# Test content
assert "What is Python?" in history
assert "Python is a programming language" in history
# Test file tracking
assert "📁 Files referenced: /home/user/main.py, /home/user/docs/readme.md" in history
assert "📁 Files referenced: /home/user/examples/" in history
# Test follow-up attribution
assert "[Gemini's Follow-up: Would you like examples?]" in history
def test_build_conversation_history_empty(self):
"""Test building history with no turns"""
test_uuid = "12345678-1234-1234-1234-123456789012"
context = ThreadContext(
thread_id=test_uuid,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:00:00Z",
tool_name="chat",
turns=[],
initial_context={},
)
history = build_conversation_history(context)
assert history == ""
class TestConversationFlow:
"""Test complete conversation flows simulating stateless MCP requests"""
@patch("utils.conversation_memory.get_redis_client")
def test_complete_conversation_cycle(self, mock_redis):
"""Test a complete 5-turn conversation until limit reached"""
mock_client = Mock()
mock_redis.return_value = mock_client
# Simulate independent MCP request cycles
# REQUEST 1: Initial request creates thread
thread_id = create_thread("chat", {"prompt": "Analyze this code"})
initial_context = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:00:00Z",
tool_name="chat",
turns=[],
initial_context={"prompt": "Analyze this code"},
)
mock_client.get.return_value = initial_context.model_dump_json()
# Add assistant response with follow-up
success = add_turn(
thread_id,
"assistant",
"Code analysis complete",
follow_up_question="Would you like me to check error handling?",
)
assert success is True
# REQUEST 2: User responds to follow-up (independent request cycle)
# Simulate retrieving updated context from Redis
context_after_1 = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:01:00Z",
tool_name="chat",
turns=[
ConversationTurn(
role="assistant",
content="Code analysis complete",
timestamp="2023-01-01T00:00:30Z",
follow_up_question="Would you like me to check error handling?",
)
],
initial_context={"prompt": "Analyze this code"},
)
mock_client.get.return_value = context_after_1.model_dump_json()
success = add_turn(thread_id, "user", "Yes, check error handling")
assert success is True
success = add_turn(
thread_id, "assistant", "Error handling reviewed", follow_up_question="Should I examine the test coverage?"
)
assert success is True
# REQUEST 3-5: Continue conversation (simulating independent cycles)
# After turn 3
context_after_3 = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:03:00Z",
tool_name="chat",
turns=[
ConversationTurn(
role="assistant",
content="Code analysis complete",
timestamp="2023-01-01T00:00:30Z",
follow_up_question="Would you like me to check error handling?",
),
ConversationTurn(role="user", content="Yes, check error handling", timestamp="2023-01-01T00:01:30Z"),
ConversationTurn(
role="assistant",
content="Error handling reviewed",
timestamp="2023-01-01T00:02:30Z",
follow_up_question="Should I examine the test coverage?",
),
],
initial_context={"prompt": "Analyze this code"},
)
mock_client.get.return_value = context_after_3.model_dump_json()
success = add_turn(thread_id, "user", "Yes, check tests")
assert success is True
success = add_turn(thread_id, "assistant", "Test coverage analyzed")
assert success is True
# REQUEST 6: Try to exceed MAX_CONVERSATION_TURNS limit - should fail
turns_at_limit = [
ConversationTurn(
role="assistant" if i % 2 == 0 else "user", content=f"Turn {i+1}", timestamp="2023-01-01T00:00:30Z"
)
for i in range(MAX_CONVERSATION_TURNS)
]
context_at_limit = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:05:00Z",
tool_name="chat",
turns=turns_at_limit,
initial_context={"prompt": "Analyze this code"},
)
mock_client.get.return_value = context_at_limit.model_dump_json()
# This should fail - conversation has reached limit
success = add_turn(thread_id, "user", "This should be rejected")
assert success is False # CONVERSATION STOPS HERE
@patch("utils.conversation_memory.get_redis_client")
def test_invalid_continuation_id_error(self, mock_redis):
"""Test that invalid continuation IDs raise proper error for restart"""
from server import reconstruct_thread_context
mock_client = Mock()
mock_redis.return_value = mock_client
mock_client.get.return_value = None # Thread not found
arguments = {"continuation_id": "invalid-uuid-12345", "prompt": "Continue conversation"}
# Should raise ValueError asking to restart
with pytest.raises(ValueError) as exc_info:
import asyncio
asyncio.run(reconstruct_thread_context(arguments))
error_msg = str(exc_info.value)
assert "Conversation thread 'invalid-uuid-12345' was not found or has expired" in error_msg
assert (
"Please restart the conversation by providing your full question/prompt without the continuation_id"
in error_msg
)
def test_dynamic_max_turns_configuration(self):
"""Test that all functions respect MAX_CONVERSATION_TURNS configuration"""
# This test ensures if we change MAX_CONVERSATION_TURNS, everything updates
# Test with different max values by patching the constant
test_values = [3, 7, 10]
for test_max in test_values:
# Create turns up to the test limit
turns = [
ConversationTurn(role="user", content=f"Turn {i}", timestamp="2023-01-01T00:00:00Z")
for i in range(test_max)
]
# Test history building respects the limit
test_uuid = "12345678-1234-1234-1234-123456789012"
context = ThreadContext(
thread_id=test_uuid,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:00:00Z",
tool_name="chat",
turns=turns,
initial_context={},
)
history = build_conversation_history(context)
expected_turn_text = f"Turn {test_max}/{MAX_CONVERSATION_TURNS}"
assert expected_turn_text in history
def test_follow_up_instructions_dynamic_behavior(self):
"""Test that follow-up instructions change correctly based on turn count and max setting"""
# Test with default MAX_CONVERSATION_TURNS
max_turns = MAX_CONVERSATION_TURNS
# Test early conversation (should allow follow-ups)
early_instructions = get_follow_up_instructions(0, max_turns)
assert "FOLLOW-UP CONVERSATIONS" in early_instructions
assert f"{max_turns - 1} more exchange" in early_instructions
# Test mid conversation
mid_instructions = get_follow_up_instructions(2, max_turns)
assert "FOLLOW-UP CONVERSATIONS" in mid_instructions
assert f"{max_turns - 3} more exchange" in mid_instructions
# Test approaching limit (should stop follow-ups)
limit_instructions = get_follow_up_instructions(max_turns - 1, max_turns)
assert "Do NOT include any follow-up questions" in limit_instructions
assert "FOLLOW-UP CONVERSATIONS" not in limit_instructions
# Test at limit
at_limit_instructions = get_follow_up_instructions(max_turns, max_turns)
assert "Do NOT include any follow-up questions" in at_limit_instructions
# Test with custom max_turns to ensure dynamic behavior
custom_max = 3
custom_early = get_follow_up_instructions(0, custom_max)
assert f"{custom_max - 1} more exchange" in custom_early
custom_limit = get_follow_up_instructions(custom_max - 1, custom_max)
assert "Do NOT include any follow-up questions" in custom_limit
def test_follow_up_instructions_defaults_to_config(self):
"""Test that follow-up instructions use MAX_CONVERSATION_TURNS when max_turns not provided"""
instructions = get_follow_up_instructions(0) # No max_turns parameter
expected_remaining = MAX_CONVERSATION_TURNS - 1
assert f"{expected_remaining} more exchange" in instructions
@patch("utils.conversation_memory.get_redis_client")
def test_complete_conversation_with_dynamic_turns(self, mock_redis):
"""Test complete conversation respecting MAX_CONVERSATION_TURNS dynamically"""
mock_client = Mock()
mock_redis.return_value = mock_client
thread_id = create_thread("chat", {"prompt": "Start conversation"})
# Simulate conversation up to MAX_CONVERSATION_TURNS - 1
for turn_num in range(MAX_CONVERSATION_TURNS - 1):
# Mock context with current turns
turns = [
ConversationTurn(
role="user" if i % 2 == 0 else "assistant", content=f"Turn {i+1}", timestamp="2023-01-01T00:00:00Z"
)
for i in range(turn_num)
]
context = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:00:00Z",
tool_name="chat",
turns=turns,
initial_context={"prompt": "Start conversation"},
)
mock_client.get.return_value = context.model_dump_json()
# Should succeed
success = add_turn(thread_id, "user", f"User turn {turn_num + 1}")
assert success is True, f"Turn {turn_num + 1} should succeed"
# Now we should be at the limit - create final context
final_turns = [
ConversationTurn(
role="user" if i % 2 == 0 else "assistant", content=f"Turn {i+1}", timestamp="2023-01-01T00:00:00Z"
)
for i in range(MAX_CONVERSATION_TURNS)
]
final_context = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:00:00Z",
tool_name="chat",
turns=final_turns,
initial_context={"prompt": "Start conversation"},
)
mock_client.get.return_value = final_context.model_dump_json()
# This should fail - at the limit
success = add_turn(thread_id, "user", "This should fail")
assert success is False, f"Turn {MAX_CONVERSATION_TURNS + 1} should fail"
@patch("utils.conversation_memory.get_redis_client")
def test_conversation_with_files_and_context_preservation(self, mock_redis):
"""Test complete conversation flow with file tracking and context preservation"""
mock_client = Mock()
mock_redis.return_value = mock_client
# Start conversation with files
thread_id = create_thread("analyze", {"prompt": "Analyze this codebase", "files": ["/project/src/"]})
# Turn 1: Claude provides context with multiple files
initial_context = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:00:00Z",
tool_name="analyze",
turns=[],
initial_context={"prompt": "Analyze this codebase", "files": ["/project/src/"]},
)
mock_client.get.return_value = initial_context.model_dump_json()
# Add Gemini's response with follow-up
success = add_turn(
thread_id,
"assistant",
"I've analyzed your codebase structure.",
follow_up_question="Would you like me to examine the test coverage?",
files=["/project/src/main.py", "/project/src/utils.py"],
tool_name="analyze",
)
assert success is True
# Turn 2: Claude responds with different files
context_turn_1 = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:01:00Z",
tool_name="analyze",
turns=[
ConversationTurn(
role="assistant",
content="I've analyzed your codebase structure.",
timestamp="2023-01-01T00:00:30Z",
follow_up_question="Would you like me to examine the test coverage?",
files=["/project/src/main.py", "/project/src/utils.py"],
tool_name="analyze",
)
],
initial_context={"prompt": "Analyze this codebase", "files": ["/project/src/"]},
)
mock_client.get.return_value = context_turn_1.model_dump_json()
# User responds with test files
success = add_turn(
thread_id, "user", "Yes, check the test coverage", files=["/project/tests/", "/project/test_main.py"]
)
assert success is True
# Turn 3: Gemini analyzes tests
context_turn_2 = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:02:00Z",
tool_name="analyze",
turns=[
ConversationTurn(
role="assistant",
content="I've analyzed your codebase structure.",
timestamp="2023-01-01T00:00:30Z",
follow_up_question="Would you like me to examine the test coverage?",
files=["/project/src/main.py", "/project/src/utils.py"],
tool_name="analyze",
),
ConversationTurn(
role="user",
content="Yes, check the test coverage",
timestamp="2023-01-01T00:01:30Z",
files=["/project/tests/", "/project/test_main.py"],
),
],
initial_context={"prompt": "Analyze this codebase", "files": ["/project/src/"]},
)
mock_client.get.return_value = context_turn_2.model_dump_json()
success = add_turn(
thread_id,
"assistant",
"Test coverage analysis complete. Coverage is 85%.",
files=["/project/tests/test_utils.py", "/project/coverage.html"],
tool_name="analyze",
)
assert success is True
# Build conversation history and verify chronological file preservation
final_context = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:03:00Z",
tool_name="analyze",
turns=[
ConversationTurn(
role="assistant",
content="I've analyzed your codebase structure.",
timestamp="2023-01-01T00:00:30Z",
follow_up_question="Would you like me to examine the test coverage?",
files=["/project/src/main.py", "/project/src/utils.py"],
tool_name="analyze",
),
ConversationTurn(
role="user",
content="Yes, check the test coverage",
timestamp="2023-01-01T00:01:30Z",
files=["/project/tests/", "/project/test_main.py"],
),
ConversationTurn(
role="assistant",
content="Test coverage analysis complete. Coverage is 85%.",
timestamp="2023-01-01T00:02:30Z",
files=["/project/tests/test_utils.py", "/project/coverage.html"],
tool_name="analyze",
),
],
initial_context={"prompt": "Analyze this codebase", "files": ["/project/src/"]},
)
history = build_conversation_history(final_context)
# Verify chronological order and speaker identification
assert "--- Turn 1 (Gemini using analyze) ---" in history
assert "--- Turn 2 (Claude) ---" in history
assert "--- Turn 3 (Gemini using analyze) ---" in history
# Verify all files are preserved in chronological order
turn_1_files = "📁 Files referenced: /project/src/main.py, /project/src/utils.py"
turn_2_files = "📁 Files referenced: /project/tests/, /project/test_main.py"
turn_3_files = "📁 Files referenced: /project/tests/test_utils.py, /project/coverage.html"
assert turn_1_files in history
assert turn_2_files in history
assert turn_3_files in history
# Verify content and follow-ups
assert "I've analyzed your codebase structure." in history
assert "Yes, check the test coverage" in history
assert "Test coverage analysis complete. Coverage is 85%." in history
assert "[Gemini's Follow-up: Would you like me to examine the test coverage?]" in history
# Verify chronological ordering (turn 1 appears before turn 2, etc.)
turn_1_pos = history.find("--- Turn 1 (Gemini using analyze) ---")
turn_2_pos = history.find("--- Turn 2 (Claude) ---")
turn_3_pos = history.find("--- Turn 3 (Gemini using analyze) ---")
assert turn_1_pos < turn_2_pos < turn_3_pos
@patch("utils.conversation_memory.get_redis_client")
def test_follow_up_question_parsing_cycle(self, mock_redis):
"""Test follow-up question persistence across request cycles"""
mock_client = Mock()
mock_redis.return_value = mock_client
thread_id = "12345678-1234-1234-1234-123456789012"
# First cycle: Assistant generates follow-up
context = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:00:00Z",
tool_name="debug",
turns=[],
initial_context={"prompt": "Debug this error"},
)
mock_client.get.return_value = context.model_dump_json()
success = add_turn(
thread_id,
"assistant",
"Found potential issue in authentication",
follow_up_question="Should I examine the authentication middleware?",
)
assert success is True
# Second cycle: Retrieve conversation history
context_with_followup = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:01:00Z",
tool_name="debug",
turns=[
ConversationTurn(
role="assistant",
content="Found potential issue in authentication",
timestamp="2023-01-01T00:00:30Z",
follow_up_question="Should I examine the authentication middleware?",
)
],
initial_context={"prompt": "Debug this error"},
)
mock_client.get.return_value = context_with_followup.model_dump_json()
# Build history to verify follow-up is preserved
history = build_conversation_history(context_with_followup)
assert "Found potential issue in authentication" in history
assert "[Gemini's Follow-up: Should I examine the authentication middleware?]" in history
@patch("utils.conversation_memory.get_redis_client")
def test_stateless_request_isolation(self, mock_redis):
"""Test that each request cycle is independent but shares context via Redis"""
mock_client = Mock()
mock_redis.return_value = mock_client
# Simulate two different "processes" accessing same thread
thread_id = "12345678-1234-1234-1234-123456789012"
# Process 1: Creates thread
initial_context = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:00:00Z",
tool_name="thinkdeep",
turns=[],
initial_context={"prompt": "Think about architecture"},
)
mock_client.get.return_value = initial_context.model_dump_json()
success = add_turn(
thread_id, "assistant", "Architecture analysis", follow_up_question="Want to explore scalability?"
)
assert success is True
# Process 2: Different "request cycle" accesses same thread
context_from_redis = ThreadContext(
thread_id=thread_id,
created_at="2023-01-01T00:00:00Z",
last_updated_at="2023-01-01T00:01:00Z",
tool_name="thinkdeep",
turns=[
ConversationTurn(
role="assistant",
content="Architecture analysis",
timestamp="2023-01-01T00:00:30Z",
follow_up_question="Want to explore scalability?",
)
],
initial_context={"prompt": "Think about architecture"},
)
mock_client.get.return_value = context_from_redis.model_dump_json()
# Verify context continuity across "processes"
retrieved_context = get_thread(thread_id)
assert retrieved_context is not None
assert len(retrieved_context.turns) == 1
assert retrieved_context.turns[0].follow_up_question == "Want to explore scalability?"
if __name__ == "__main__":
pytest.main([__file__])

4
tools/analyze.py
View File

@@ -88,6 +88,10 @@ class AnalyzeTool(BaseTool):
"description": "Enable web search for documentation, best practices, and current information. Particularly useful for: brainstorming sessions, architectural design discussions, exploring industry best practices, working with specific frameworks/technologies, researching solutions to complex problems, or when current documentation and community insights would enhance the analysis.",
"default": True,
},
"continuation_id": {
"type": "string",
"description": "Thread continuation ID for multi-turn conversations. Only provide this if continuing a previous conversation thread.",
},
},
"required": ["files", "question"],
}

273
tools/base.py
View File

@@ -26,7 +26,7 @@ from pydantic import BaseModel, Field
from config import MCP_PROMPT_SIZE_LIMIT
from utils.file_utils import read_file_content, translate_path_for_environment
from .models import ClarificationRequest, ToolOutput
from .models import ClarificationRequest, ContinuationOffer, FollowUpRequest, ToolOutput
class ToolRequest(BaseModel):
@@ -50,6 +50,10 @@ class ToolRequest(BaseModel):
False,
description="Enable web search for documentation, best practices, and current information. Particularly useful for: brainstorming sessions, architectural design discussions, exploring industry best practices, working with specific frameworks/technologies, researching solutions to complex problems, or when current documentation and community insights would enhance the analysis.",
)
continuation_id: Optional[str] = Field(
None,
description="Thread continuation ID for multi-turn conversations. Only provide this if continuing a previous conversation thread.",
)
class BaseTool(ABC):
@@ -375,6 +379,20 @@ If any of these would strengthen your analysis, specify what Claude should searc
# This is delegated to the tool implementation for customization
prompt = await self.prepare_prompt(request)
# Add follow-up instructions for new conversations (not threaded)
continuation_id = getattr(request, "continuation_id", None)
if not continuation_id:
# Import here to avoid circular imports
from server import get_follow_up_instructions
import logging
follow_up_instructions = get_follow_up_instructions(0) # New conversation, turn 0
prompt = f"{prompt}\n\n{follow_up_instructions}"
logging.debug(f"Added follow-up instructions for new {self.name} conversation")
else:
import logging
logging.debug(f"Continuing {self.name} conversation with thread {continuation_id}")
# Extract model configuration from request or use defaults
from config import GEMINI_MODEL
@@ -425,10 +443,10 @@ If any of these would strengthen your analysis, specify what Claude should searc
def _parse_response(self, raw_text: str, request) -> ToolOutput:
"""
Parse the raw response and determine if it's a clarification request.
Parse the raw response and determine if it's a clarification request or follow-up.
Some tools may return JSON indicating they need more information.
This method detects such responses and formats them appropriately.
Some tools may return JSON indicating they need more information or want to
continue the conversation. This method detects such responses and formats them.
Args:
raw_text: The raw text response from the model
@@ -437,6 +455,15 @@ If any of these would strengthen your analysis, specify what Claude should searc
Returns:
ToolOutput: Standardized output object
"""
# Check for follow-up questions in JSON blocks at the end of the response
follow_up_question = self._extract_follow_up_question(raw_text)
import logging
if follow_up_question:
logging.debug(f"Found follow-up question in {self.name} response: {follow_up_question.get('follow_up_question', 'N/A')}")
else:
logging.debug(f"No follow-up question found in {self.name} response")
try:
# Try to parse as JSON to check for clarification requests
potential_json = json.loads(raw_text.strip())
@@ -460,6 +487,20 @@ If any of these would strengthen your analysis, specify what Claude should searc
# Normal text response - format using tool-specific formatting
formatted_content = self.format_response(raw_text, request)
# If we found a follow-up question, prepare the threading response
if follow_up_question:
return self._create_follow_up_response(formatted_content, follow_up_question, request)
# Check if we should offer Claude a continuation opportunity
continuation_offer = self._check_continuation_opportunity(request)
import logging
if continuation_offer:
logging.debug(f"Creating continuation offer for {self.name} with {continuation_offer['remaining_turns']} turns remaining")
return self._create_continuation_offer_response(formatted_content, continuation_offer, request)
else:
logging.debug(f"No continuation offer created for {self.name}")
# Determine content type based on the formatted content
content_type = (
"markdown" if any(marker in formatted_content for marker in ["##", "**", "`", "- ", "1. "]) else "text"
@@ -472,6 +513,230 @@ If any of these would strengthen your analysis, specify what Claude should searc
metadata={"tool_name": self.name},
)
def _extract_follow_up_question(self, text: str) -> Optional[dict]:
"""
Extract follow-up question from JSON blocks in the response.
Looks for JSON blocks containing follow_up_question at the end of responses.
Args:
text: The response text to parse
Returns:
Dict with follow-up data if found, None otherwise
"""
import re
# Look for JSON blocks that contain follow_up_question
# Pattern handles optional leading whitespace and indentation
json_pattern = r'```json\s*\n\s*(\{.*?"follow_up_question".*?\})\s*\n\s*```'
matches = re.findall(json_pattern, text, re.DOTALL)
if not matches:
return None
# Take the last match (most recent follow-up)
try:
# Clean up the JSON string - remove excess whitespace and normalize
json_str = re.sub(r"\n\s+", "\n", matches[-1]).strip()
follow_up_data = json.loads(json_str)
if "follow_up_question" in follow_up_data:
return follow_up_data
except (json.JSONDecodeError, ValueError):
pass
return None
def _create_follow_up_response(self, content: str, follow_up_data: dict, request) -> ToolOutput:
"""
Create a response with follow-up question for conversation threading.
Args:
content: The main response content
follow_up_data: Dict containing follow_up_question and optional suggested_params
request: Original request for context
Returns:
ToolOutput configured for conversation continuation
"""
from utils.conversation_memory import add_turn, create_thread
# Create or get thread ID
continuation_id = getattr(request, "continuation_id", None)
if continuation_id:
# This is a continuation - add this turn to existing thread
request_files = getattr(request, "files", []) or []
success = add_turn(
continuation_id,
"assistant",
content,
follow_up_question=follow_up_data.get("follow_up_question"),
files=request_files,
tool_name=self.name,
)
if not success:
# Thread not found or at limit, return normal response
return ToolOutput(
status="success",
content=content,
content_type="markdown",
metadata={"tool_name": self.name},
)
thread_id = continuation_id
else:
# Create new thread
try:
thread_id = create_thread(
tool_name=self.name, initial_request=request.model_dump() if hasattr(request, "model_dump") else {}
)
# Add the assistant's response with follow-up
request_files = getattr(request, "files", []) or []
add_turn(
thread_id,
"assistant",
content,
follow_up_question=follow_up_data.get("follow_up_question"),
files=request_files,
tool_name=self.name,
)
except Exception as e:
# Threading failed, return normal response
import logging
logging.warning(f"Follow-up threading failed in {self.name}: {str(e)}")
return ToolOutput(
status="success",
content=content,
content_type="markdown",
metadata={"tool_name": self.name, "follow_up_error": str(e)},
)
# Create follow-up request
follow_up_request = FollowUpRequest(
continuation_id=thread_id,
question_to_user=follow_up_data["follow_up_question"],
suggested_tool_params=follow_up_data.get("suggested_params"),
ui_hint=follow_up_data.get("ui_hint"),
)
# Strip the JSON block from the content since it's now in the follow_up_request
clean_content = self._remove_follow_up_json(content)
return ToolOutput(
status="requires_continuation",
content=clean_content,
content_type="markdown",
follow_up_request=follow_up_request,
metadata={"tool_name": self.name, "thread_id": thread_id},
)
def _remove_follow_up_json(self, text: str) -> str:
"""Remove follow-up JSON blocks from the response text"""
import re
# Remove JSON blocks containing follow_up_question
pattern = r'```json\s*\n\s*\{.*?"follow_up_question".*?\}\s*\n\s*```'
return re.sub(pattern, "", text, flags=re.DOTALL).strip()
def _check_continuation_opportunity(self, request) -> Optional[dict]:
"""
Check if we should offer Claude a continuation opportunity.
This is called when Gemini doesn't ask a follow-up question, but we want
to give Claude the chance to continue the conversation if needed.
Args:
request: The original request
Returns:
Dict with continuation data if opportunity should be offered, None otherwise
"""
# Only offer continuation for new conversations (not already threaded)
continuation_id = getattr(request, "continuation_id", None)
if continuation_id:
# This is already a threaded conversation, don't offer continuation
# (either Gemini will ask follow-up or conversation naturally ends)
return None
# Only offer if we haven't reached conversation limits
try:
from utils.conversation_memory import MAX_CONVERSATION_TURNS
# For new conversations, we have MAX_CONVERSATION_TURNS - 1 remaining
# (since this response will be turn 1)
remaining_turns = MAX_CONVERSATION_TURNS - 1
if remaining_turns <= 0:
return None
# Offer continuation opportunity
return {"remaining_turns": remaining_turns, "tool_name": self.name}
except Exception:
# If anything fails, don't offer continuation
return None
def _create_continuation_offer_response(self, content: str, continuation_data: dict, request) -> ToolOutput:
"""
Create a response offering Claude the opportunity to continue conversation.
Args:
content: The main response content
continuation_data: Dict containing remaining_turns and tool_name
request: Original request for context
Returns:
ToolOutput configured with continuation offer
"""
from utils.conversation_memory import create_thread
try:
# Create new thread for potential continuation
thread_id = create_thread(
tool_name=self.name, initial_request=request.model_dump() if hasattr(request, "model_dump") else {}
)
# Add this response as the first turn (assistant turn)
from utils.conversation_memory import add_turn
request_files = getattr(request, "files", []) or []
add_turn(thread_id, "assistant", content, files=request_files, tool_name=self.name)
# Create continuation offer
remaining_turns = continuation_data["remaining_turns"]
continuation_offer = ContinuationOffer(
continuation_id=thread_id,
message_to_user=(
f"If you'd like to continue this analysis or need further details, "
f"you can use the continuation_id '{thread_id}' in your next {self.name} tool call. "
f"You have {remaining_turns} more exchange(s) available in this conversation thread."
),
suggested_tool_params={
"continuation_id": thread_id,
"prompt": "[Your follow-up question or request for additional analysis]",
},
remaining_turns=remaining_turns,
)
return ToolOutput(
status="continuation_available",
content=content,
content_type="markdown",
continuation_offer=continuation_offer,
metadata={"tool_name": self.name, "thread_id": thread_id, "remaining_turns": remaining_turns},
)
except Exception as e:
# If threading fails, return normal response but log the error
import logging
logging.warning(f"Conversation threading failed in {self.name}: {str(e)}")
return ToolOutput(
status="success",
content=content,
content_type="markdown",
metadata={"tool_name": self.name, "threading_error": str(e)},
)
@abstractmethod
async def prepare_prompt(self, request) -> str:
"""

4
tools/chat.py
View File

@@ -73,6 +73,10 @@ class ChatTool(BaseTool):
"description": "Enable web search for documentation, best practices, and current information. Particularly useful for: brainstorming sessions, architectural design discussions, exploring industry best practices, working with specific frameworks/technologies, researching solutions to complex problems, or when current documentation and community insights would enhance the analysis.",
"default": True,
},
"continuation_id": {
"type": "string",
"description": "Thread continuation ID for multi-turn conversations. Only provide this if continuing a previous conversation thread.",
},
},
"required": ["prompt"],
}

4
tools/codereview.py
View File

@@ -126,6 +126,10 @@ class CodeReviewTool(BaseTool):
"description": "Enable web search for documentation, best practices, and current information. Particularly useful for: brainstorming sessions, architectural design discussions, exploring industry best practices, working with specific frameworks/technologies, researching solutions to complex problems, or when current documentation and community insights would enhance the analysis.",
"default": True,
},
"continuation_id": {
"type": "string",
"description": "Thread continuation ID for multi-turn conversations. Only provide this if continuing a previous conversation thread.",
},
},
"required": ["files", "context"],
}

4
tools/debug.py
View File

@@ -91,6 +91,10 @@ class DebugIssueTool(BaseTool):
"description": "Enable web search for documentation, best practices, and current information. Particularly useful for: brainstorming sessions, architectural design discussions, exploring industry best practices, working with specific frameworks/technologies, researching solutions to complex problems, or when current documentation and community insights would enhance the analysis.",
"default": True,
},
"continuation_id": {
"type": "string",
"description": "Thread continuation ID for multi-turn conversations. Only provide this if continuing a previous conversation thread.",
},
},
"required": ["error_description"],
}

41
tools/models.py
View File

@@ -7,13 +7,50 @@ from typing import Any, Literal, Optional
from pydantic import BaseModel, Field
class FollowUpRequest(BaseModel):
"""Request for follow-up conversation turn"""
continuation_id: str = Field(..., description="Thread continuation ID for multi-turn conversations")
question_to_user: str = Field(..., description="Follow-up question to ask Claude")
suggested_tool_params: Optional[dict[str, Any]] = Field(
None, description="Suggested parameters for the next tool call"
)
ui_hint: Optional[str] = Field(
None, description="UI hint for Claude (e.g., 'text_input', 'file_select', 'multi_choice')"
)
class ContinuationOffer(BaseModel):
"""Offer for Claude to continue conversation when Gemini doesn't ask follow-up"""
continuation_id: str = Field(..., description="Thread continuation ID for multi-turn conversations")
message_to_user: str = Field(..., description="Message explaining continuation opportunity to Claude")
suggested_tool_params: Optional[dict[str, Any]] = Field(
None, description="Suggested parameters for continued tool usage"
)
remaining_turns: int = Field(..., description="Number of conversation turns remaining")
class ToolOutput(BaseModel):
"""Standardized output format for all tools"""
status: Literal["success", "error", "requires_clarification", "requires_file_prompt"] = "success"
content: str = Field(..., description="The main content/response from the tool")
status: Literal[
"success",
"error",
"requires_clarification",
"requires_file_prompt",
"requires_continuation",
"continuation_available",
] = "success"
content: Optional[str] = Field(None, description="The main content/response from the tool")
content_type: Literal["text", "markdown", "json"] = "text"
metadata: Optional[dict[str, Any]] = Field(default_factory=dict)
follow_up_request: Optional[FollowUpRequest] = Field(
None, description="Optional follow-up request for continued conversation"
)
continuation_offer: Optional[ContinuationOffer] = Field(
None, description="Optional offer for Claude to continue conversation"
)
class ClarificationRequest(BaseModel):

6
tools/precommit.py
View File

@@ -100,6 +100,12 @@ class Precommit(BaseTool):
"description": "Enable web search for documentation, best practices, and current information. Particularly useful for: brainstorming sessions, architectural design discussions, exploring industry best practices, working with specific frameworks/technologies, researching solutions to complex problems, or when current documentation and community insights would enhance the analysis.",
"default": True,
}
# Add continuation_id parameter
if "properties" in schema and "continuation_id" not in schema["properties"]:
schema["properties"]["continuation_id"] = {
"type": "string",
"description": "Thread continuation ID for multi-turn conversations. Only provide this if continuing a previous conversation thread.",
}
return schema
def get_system_prompt(self) -> str:

4
tools/thinkdeep.py
View File

@@ -87,6 +87,10 @@ class ThinkDeepTool(BaseTool):
"description": "Enable web search for documentation, best practices, and current information. Particularly useful for: brainstorming sessions, architectural design discussions, exploring industry best practices, working with specific frameworks/technologies, researching solutions to complex problems, or when current documentation and community insights would enhance the analysis.",
"default": True,
},
"continuation_id": {
"type": "string",
"description": "Thread continuation ID for multi-turn conversations. Only provide this if continuing a previous conversation thread.",
},
},
"required": ["current_analysis"],
}

195
utils/conversation_memory.py Normal file
View File

@@ -0,0 +1,195 @@
"""
Conversation Memory for AI-to-AI Multi-turn Discussions
This module provides conversation persistence and context reconstruction for
stateless MCP environments. It enables multi-turn conversations between Claude
and Gemini by storing conversation state in Redis across independent request cycles.
Key Features:
- UUID-based conversation thread identification
- Turn-by-turn conversation history storage
- Automatic turn limiting to prevent runaway conversations
- Context reconstruction for stateless request continuity
- Redis-based persistence with automatic expiration
"""
import os
import uuid
from datetime import datetime, timezone
from typing import Any, Optional
from pydantic import BaseModel
# Configuration constants
MAX_CONVERSATION_TURNS = 5 # Maximum turns allowed per conversation thread
class ConversationTurn(BaseModel):
"""Single turn in a conversation"""
role: str # "user" or "assistant"
content: str
timestamp: str
follow_up_question: Optional[str] = None
files: Optional[list[str]] = None # Files referenced in this turn
tool_name: Optional[str] = None # Tool used for this turn
class ThreadContext(BaseModel):
"""Complete conversation context"""
thread_id: str
created_at: str
last_updated_at: str
tool_name: str
turns: list[ConversationTurn]
initial_context: dict[str, Any]
def get_redis_client():
"""Get Redis client from environment"""
try:
import redis
redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
return redis.from_url(redis_url, decode_responses=True)
except ImportError:
raise ValueError("redis package required. Install with: pip install redis")
def create_thread(tool_name: str, initial_request: dict[str, Any]) -> str:
"""Create new conversation thread and return thread ID"""
thread_id = str(uuid.uuid4())
now = datetime.now(timezone.utc).isoformat()
# Filter out non-serializable parameters
filtered_context = {
k: v
for k, v in initial_request.items()
if k not in ["temperature", "thinking_mode", "model", "continuation_id"]
}
context = ThreadContext(
thread_id=thread_id,
created_at=now,
last_updated_at=now,
tool_name=tool_name,
turns=[],
initial_context=filtered_context,
)
# Store in Redis with 1 hour TTL
client = get_redis_client()
key = f"thread:{thread_id}"
client.setex(key, 3600, context.model_dump_json())
return thread_id
def get_thread(thread_id: str) -> Optional[ThreadContext]:
"""Retrieve thread context from Redis"""
if not thread_id or not _is_valid_uuid(thread_id):
return None
try:
client = get_redis_client()
key = f"thread:{thread_id}"
data = client.get(key)
if data:
return ThreadContext.model_validate_json(data)
return None
except Exception:
return None
def add_turn(
thread_id: str,
role: str,
content: str,
follow_up_question: Optional[str] = None,
files: Optional[list[str]] = None,
tool_name: Optional[str] = None,
) -> bool:
"""Add turn to existing thread"""
context = get_thread(thread_id)
if not context:
return False
# Check turn limit
if len(context.turns) >= MAX_CONVERSATION_TURNS:
return False
# Add new turn
turn = ConversationTurn(
role=role,
content=content,
timestamp=datetime.now(timezone.utc).isoformat(),
follow_up_question=follow_up_question,
files=files,
tool_name=tool_name,
)
context.turns.append(turn)
context.last_updated_at = datetime.now(timezone.utc).isoformat()
# Save back to Redis
try:
client = get_redis_client()
key = f"thread:{thread_id}"
client.setex(key, 3600, context.model_dump_json()) # Refresh TTL
return True
except Exception:
return False
def build_conversation_history(context: ThreadContext) -> str:
"""Build formatted conversation history"""
if not context.turns:
return ""
history_parts = [
"=== CONVERSATION HISTORY ===",
f"Thread: {context.thread_id}",
f"Tool: {context.tool_name}",
f"Turn {len(context.turns)}/{MAX_CONVERSATION_TURNS}",
"",
"Previous exchanges:",
]
for i, turn in enumerate(context.turns, 1):
role_label = "Claude" if turn.role == "user" else "Gemini"
# Add turn header with tool info if available
turn_header = f"\n--- Turn {i} ({role_label}"
if turn.tool_name:
turn_header += f" using {turn.tool_name}"
turn_header += ") ---"
history_parts.append(turn_header)
# Add files context if present
if turn.files:
history_parts.append(f"📁 Files referenced: {', '.join(turn.files)}")
history_parts.append("") # Empty line for readability
# Add the actual content
history_parts.append(turn.content)
# Add follow-up question if present
if turn.follow_up_question:
history_parts.append(f"\n[Gemini's Follow-up: {turn.follow_up_question}]")
history_parts.extend(
["", "=== END HISTORY ===", "", "Continue this conversation by building on the previous context."]
)
return "\n".join(history_parts)
def _is_valid_uuid(val: str) -> bool:
"""Validate UUID format for security"""
try:
uuid.UUID(val)
return True
except ValueError:
return False

14
utils/file_utils.py
View File

@@ -341,6 +341,20 @@ def expand_paths(paths: list[str], extensions: Optional[set[str]] = None) -> lis
if not path_obj.exists():
continue
# Safety check: Prevent reading entire home directory or workspace root
# This could expose too many files and cause performance issues
if path_obj.is_dir():
resolved_project_root = PROJECT_ROOT.resolve()
resolved_path = path_obj.resolve()
# Check if this is the entire project root/home directory
if resolved_path == resolved_project_root:
logger.warning(
f"Ignoring request to read entire project root directory: {path}. "
f"Please specify individual files or subdirectories instead."
)
continue
if path_obj.is_file():
# Add file directly
if str(path_obj) not in seen: