This is page 3 of 6. Use http://codebase.md/tosin2013/mcp-codebase-insight?page={x} to view the full context.
# Directory Structure
```
├── .bumpversion.cfg
├── .codecov.yml
├── .compile-venv-py3.11
│ ├── bin
│ │ ├── activate
│ │ ├── activate.csh
│ │ ├── activate.fish
│ │ ├── Activate.ps1
│ │ ├── coverage
│ │ ├── coverage-3.11
│ │ ├── coverage3
│ │ ├── pip
│ │ ├── pip-compile
│ │ ├── pip-sync
│ │ ├── pip3
│ │ ├── pip3.11
│ │ ├── py.test
│ │ ├── pyproject-build
│ │ ├── pytest
│ │ ├── python
│ │ ├── python3
│ │ ├── python3.11
│ │ └── wheel
│ └── pyvenv.cfg
├── .env.example
├── .github
│ └── workflows
│ ├── build-verification.yml
│ ├── publish.yml
│ └── tdd-verification.yml
├── .gitignore
├── async_fixture_wrapper.py
├── CHANGELOG.md
├── CLAUDE.md
├── codebase_structure.txt
├── component_test_runner.py
├── CONTRIBUTING.md
├── core_workflows.txt
├── debug_tests.md
├── Dockerfile
├── docs
│ ├── adrs
│ │ └── 001_use_docker_for_qdrant.md
│ ├── api.md
│ ├── components
│ │ └── README.md
│ ├── cookbook.md
│ ├── development
│ │ ├── CODE_OF_CONDUCT.md
│ │ ├── CONTRIBUTING.md
│ │ └── README.md
│ ├── documentation_map.md
│ ├── documentation_summary.md
│ ├── features
│ │ ├── adr-management.md
│ │ ├── code-analysis.md
│ │ └── documentation.md
│ ├── getting-started
│ │ ├── configuration.md
│ │ ├── docker-setup.md
│ │ ├── installation.md
│ │ ├── qdrant_setup.md
│ │ └── quickstart.md
│ ├── qdrant_setup.md
│ ├── README.md
│ ├── SSE_INTEGRATION.md
│ ├── system_architecture
│ │ └── README.md
│ ├── templates
│ │ └── adr.md
│ ├── testing_guide.md
│ ├── troubleshooting
│ │ ├── common-issues.md
│ │ └── faq.md
│ ├── vector_store_best_practices.md
│ └── workflows
│ └── README.md
├── error_logs.txt
├── examples
│ └── use_with_claude.py
├── github-actions-documentation.md
├── Makefile
├── module_summaries
│ ├── backend_summary.txt
│ ├── database_summary.txt
│ └── frontend_summary.txt
├── output.txt
├── package-lock.json
├── package.json
├── PLAN.md
├── prepare_codebase.sh
├── PULL_REQUEST.md
├── pyproject.toml
├── pytest.ini
├── README.md
├── requirements-3.11.txt
├── requirements-3.11.txt.backup
├── requirements-dev.txt
├── requirements.in
├── requirements.txt
├── run_build_verification.sh
├── run_fixed_tests.sh
├── run_test_with_path_fix.sh
├── run_tests.py
├── scripts
│ ├── check_qdrant_health.sh
│ ├── compile_requirements.sh
│ ├── load_example_patterns.py
│ ├── macos_install.sh
│ ├── README.md
│ ├── setup_qdrant.sh
│ ├── start_mcp_server.sh
│ ├── store_code_relationships.py
│ ├── store_report_in_mcp.py
│ ├── validate_knowledge_base.py
│ ├── validate_poc.py
│ ├── validate_vector_store.py
│ └── verify_build.py
├── server.py
├── setup_qdrant_collection.py
├── setup.py
├── src
│ └── mcp_codebase_insight
│ ├── __init__.py
│ ├── __main__.py
│ ├── asgi.py
│ ├── core
│ │ ├── __init__.py
│ │ ├── adr.py
│ │ ├── cache.py
│ │ ├── component_status.py
│ │ ├── config.py
│ │ ├── debug.py
│ │ ├── di.py
│ │ ├── documentation.py
│ │ ├── embeddings.py
│ │ ├── errors.py
│ │ ├── health.py
│ │ ├── knowledge.py
│ │ ├── metrics.py
│ │ ├── prompts.py
│ │ ├── sse.py
│ │ ├── state.py
│ │ ├── task_tracker.py
│ │ ├── tasks.py
│ │ └── vector_store.py
│ ├── models.py
│ ├── server_test_isolation.py
│ ├── server.py
│ ├── utils
│ │ ├── __init__.py
│ │ └── logger.py
│ └── version.py
├── start-mcpserver.sh
├── summary_document.txt
├── system-architecture.md
├── system-card.yml
├── test_fix_helper.py
├── test_fixes.md
├── test_function.txt
├── test_imports.py
├── tests
│ ├── components
│ │ ├── conftest.py
│ │ ├── test_core_components.py
│ │ ├── test_embeddings.py
│ │ ├── test_knowledge_base.py
│ │ ├── test_sse_components.py
│ │ ├── test_stdio_components.py
│ │ ├── test_task_manager.py
│ │ └── test_vector_store.py
│ ├── config
│ │ └── test_config_and_env.py
│ ├── conftest.py
│ ├── integration
│ │ ├── fixed_test2.py
│ │ ├── test_api_endpoints.py
│ │ ├── test_api_endpoints.py-e
│ │ ├── test_communication_integration.py
│ │ └── test_server.py
│ ├── README.md
│ ├── README.test.md
│ ├── test_build_verifier.py
│ └── test_file_relationships.py
└── trajectories
└── tosinakinosho
├── anthropic_filemap__claude-3-sonnet-20240229__t-0.00__p-1.00__c-3.00___db62b9
│ └── db62b9
│ └── config.yaml
├── default__claude-3-5-sonnet-20240620__t-0.00__p-1.00__c-3.00___03565e
│ └── 03565e
│ ├── 03565e.traj
│ └── config.yaml
└── default__openrouter
└── anthropic
└── claude-3.5-sonnet-20240620:beta__t-0.00__p-1.00__c-3.00___03565e
└── 03565e
├── 03565e.pred
├── 03565e.traj
└── config.yaml
```
# Files
--------------------------------------------------------------------------------
/docs/cookbook.md:
--------------------------------------------------------------------------------
```markdown
# MCP Codebase Insight Cookbook
This cookbook provides practical examples, common use cases, and solutions for working with the MCP Codebase Insight system. Each recipe includes step-by-step instructions, code examples, and explanations.
## Table of Contents
- [Setup and Configuration](#setup-and-configuration)
- [Vector Store Operations](#vector-store-operations)
- [Code Analysis](#code-analysis)
- [Knowledge Base Integration](#knowledge-base-integration)
- [Task Management](#task-management)
- [Transport Protocol Usage](#transport-protocol-usage)
- [Troubleshooting](#troubleshooting)
## Setup and Configuration
### Recipe: Quick Start Setup
```bash
# 1. Clone the repository
git clone https://github.com/your-org/mcp-codebase-insight.git
cd mcp-codebase-insight
# 2. Create and activate virtual environment
python -m venv .venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
# 3. Install dependencies
pip install -r requirements.txt
# 4. Set up environment variables
cp .env.example .env
# Edit .env with your configuration
```
### Recipe: Configure Vector Store
```python
from mcp_codebase_insight.core.vector_store import VectorStore
from mcp_codebase_insight.core.embeddings import SentenceTransformerEmbedding
async def setup_vector_store():
# Initialize embedder
embedder = SentenceTransformerEmbedding(
model_name="sentence-transformers/all-MiniLM-L6-v2"
)
await embedder.initialize()
# Initialize vector store
vector_store = VectorStore(
url="http://localhost:6333",
embedder=embedder,
collection_name="mcp-codebase-insight",
api_key="your-api-key", # Optional
vector_name="default"
)
await vector_store.initialize()
return vector_store
```
## Vector Store Operations
### Recipe: Store and Search Code Snippets
```python
async def store_code_snippet(vector_store, code: str, metadata: dict):
await vector_store.add_vector(
text=code,
metadata={
"type": "code",
"content": code,
**metadata
}
)
async def search_similar_code(vector_store, query: str, limit: int = 5):
results = await vector_store.search_similar(
query=query,
limit=limit
)
return results
# Usage example
code_snippet = """
def calculate_sum(a: int, b: int) -> int:
return a + b
"""
metadata = {
"filename": "math_utils.py",
"function_name": "calculate_sum",
"language": "python"
}
await store_code_snippet(vector_store, code_snippet, metadata)
similar_snippets = await search_similar_code(vector_store, "function to add two numbers")
```
### Recipe: Batch Processing Code Files
```python
import asyncio
from pathlib import Path
async def process_codebase(vector_store, root_dir: str):
async def process_file(file_path: Path):
if not file_path.suffix == '.py': # Adjust for your needs
return
code = file_path.read_text()
await store_code_snippet(vector_store, code, {
"filename": file_path.name,
"path": str(file_path),
"language": "python"
})
root = Path(root_dir)
tasks = [
process_file(f)
for f in root.rglob('*')
if f.is_file()
]
await asyncio.gather(*tasks)
```
## Code Analysis
### Recipe: Detect Architectural Patterns
```python
from mcp_codebase_insight.analysis.patterns import PatternDetector
async def analyze_architecture(code_path: str):
detector = PatternDetector()
patterns = await detector.detect_patterns(code_path)
for pattern in patterns:
print(f"Pattern: {pattern.name}")
print(f"Location: {pattern.location}")
print(f"Confidence: {pattern.confidence}")
print("---")
```
### Recipe: Generate Code Insights
```python
from mcp_codebase_insight.analysis.insights import InsightGenerator
async def generate_insights(vector_store, codebase_path: str):
generator = InsightGenerator(vector_store)
insights = await generator.analyze_codebase(codebase_path)
return {
"complexity_metrics": insights.complexity,
"dependency_graph": insights.dependencies,
"architectural_patterns": insights.patterns,
"recommendations": insights.recommendations
}
```
## Knowledge Base Integration
### Recipe: Store and Query Documentation
```python
from mcp_codebase_insight.kb.store import KnowledgeBase
async def manage_documentation(kb: KnowledgeBase):
# Store documentation
await kb.store_document(
content="API documentation content...",
metadata={
"type": "api_doc",
"version": "1.0",
"category": "reference"
}
)
# Query documentation
results = await kb.search(
query="How to configure authentication",
filters={
"type": "api_doc",
"category": "reference"
}
)
```
## Task Management
### Recipe: Create and Track Tasks
```python
from mcp_codebase_insight.tasks.manager import TaskManager
async def manage_tasks(task_manager: TaskManager):
# Create a new task
task = await task_manager.create_task(
title="Implement authentication",
description="Add OAuth2 authentication to API endpoints",
priority="high",
tags=["security", "api"]
)
# Update task status
await task_manager.update_task(
task_id=task.id,
status="in_progress",
progress=0.5
)
# Query tasks
active_tasks = await task_manager.get_tasks(
filters={
"status": "in_progress",
"tags": ["security"]
}
)
```
## Transport Protocol Usage
### Recipe: Using SSE Transport
```python
from mcp_codebase_insight.transport.sse import SSETransport
async def setup_sse():
transport = SSETransport(
url="http://localhost:8000/events",
headers={"Authorization": "Bearer your-token"}
)
async with transport:
await transport.subscribe("codebase_updates")
async for event in transport.events():
print(f"Received update: {event.data}")
```
### Recipe: Using StdIO Transport
```python
from mcp_codebase_insight.transport.stdio import StdIOTransport
async def use_stdio():
transport = StdIOTransport()
async with transport:
# Send command
await transport.send_command({
"type": "analyze",
"payload": {"path": "src/main.py"}
})
# Receive response
response = await transport.receive_response()
print(f"Analysis result: {response}")
```
## Troubleshooting
### Recipe: Validate Vector Store Health
```python
async def check_vector_store_health(config: dict) -> bool:
try:
# Initialize components
embedder = SentenceTransformerEmbedding(
model_name="sentence-transformers/all-MiniLM-L6-v2"
)
await embedder.initialize()
vector_store = VectorStore(
url=config["QDRANT_URL"],
embedder=embedder,
collection_name=config["COLLECTION_NAME"]
)
await vector_store.initialize()
# Test basic operations
test_text = "def test_function():\n pass"
await vector_store.add_vector(
text=test_text,
metadata={"type": "test"}
)
results = await vector_store.search_similar(
query=test_text,
limit=1
)
return len(results) > 0
except Exception as e:
print(f"Health check failed: {e}")
return False
```
### Recipe: Debug Transport Issues
```python
import logging
from mcp_codebase_insight.transport.debug import TransportDebugger
async def debug_transport_issues():
# Enable detailed logging
logging.basicConfig(level=logging.DEBUG)
debugger = TransportDebugger()
# Test SSE connection
sse_status = await debugger.check_sse_connection(
url="http://localhost:8000/events"
)
print(f"SSE Status: {sse_status}")
# Test StdIO communication
stdio_status = await debugger.check_stdio_communication()
print(f"StdIO Status: {stdio_status}")
# Generate diagnostic report
report = await debugger.generate_diagnostic_report()
print(report)
```
## Best Practices
1. Always use async/await when working with the system's async functions
2. Initialize components in a context manager or properly handle cleanup
3. Use structured error handling for vector store operations
4. Implement retry logic for network-dependent operations
5. Cache frequently accessed vector embeddings
6. Use batch operations when processing multiple items
7. Implement proper logging for debugging
8. Regular health checks for system components
## Common Issues and Solutions
1. **Vector Store Connection Issues**
- Check if Qdrant is running and accessible
- Verify API key if authentication is enabled
- Ensure proper network connectivity
2. **Embedding Generation Failures**
- Verify model availability and access
- Check input text formatting
- Monitor memory usage for large inputs
3. **Transport Protocol Errors**
- Verify endpoint URLs and authentication
- Check for firewall or proxy issues
- Monitor connection timeouts
4. **Performance Issues**
- Use batch operations for multiple items
- Implement caching where appropriate
- Monitor and optimize vector store queries
For more detailed information, refer to the [official documentation](docs/README.md) and [API reference](docs/api-reference.md).
```
--------------------------------------------------------------------------------
/trajectories/tosinakinosho/anthropic_filemap__claude-3-sonnet-20240229__t-0.00__p-1.00__c-3.00___db62b9/db62b9/config.yaml:
--------------------------------------------------------------------------------
```yaml
'{"env":{"deployment":{"image":"python:3.11","port":null,"docker_args":[],"startup_timeout":180.0,"pull":"missing","remove_images":false,"python_standalone_dir":"/root","platform":null,"type":"docker"},"repo":{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight","base_commit":"HEAD","type":"local"},"post_startup_commands":[],"post_startup_command_timeout":500,"name":"main"},"agent":{"name":"main","templates":{"system_template":"You
are a helpful assistant that can interact with a computer to solve tasks.","instance_template":"<uploaded_files>\n{{working_dir}}\n</uploaded_files>\nI''ve
uploaded a python code repository in the directory {{working_dir}}. Consider the
following PR description:\n\n<pr_description>\n{{problem_statement}}\n</pr_description>\n\nCan
you help me implement the necessary changes to the repository so that the requirements
specified in the <pr_description> are met?\nI''ve already taken care of all changes
to any of the test files described in the <pr_description>. This means you DON''T
have to modify the testing logic or any of the tests in any way!\nYour task is to
make the minimal changes to non-tests files in the {{working_dir}} directory to
ensure the <pr_description> is satisfied.\nFollow these steps to resolve the issue:\n1.
As a first step, it might be a good idea to find and read code relevant to the <pr_description>\n2.
Create a script to reproduce the error and execute it with `python <filename.py>`
using the bash tool, to confirm the error\n3. Edit the sourcecode of the repo to
resolve the issue\n4. Rerun your reproduce script and confirm that the error is
fixed!\n5. Think about edgecases and make sure your fix handles them as well\nYour
thinking should be thorough and so it''s fine if it''s very long.","next_step_template":"OBSERVATION:\n{{observation}}","next_step_truncated_observation_template":"Observation:
{{observation}}<response clipped><NOTE>Observations should not exceeded {{max_observation_length}}
characters. {{elided_chars}} characters were elided. Please try a different command
that produces less output or use head/tail/grep/redirect the output to a file. Do
not use interactive pagers.</NOTE>","max_observation_length":100000,"next_step_no_output_template":"Your
command ran successfully and did not produce any output.","strategy_template":null,"demonstration_template":null,"demonstrations":[],"put_demos_in_history":false,"shell_check_error_template":"Your
bash command contained syntax errors and was NOT executed. Please fix the syntax
errors and try again. This can be the result of not adhering to the syntax for multi-line
commands. Here is the output of `bash -n`:\n{{bash_stdout}}\n{{bash_stderr}}","command_cancelled_timeout_template":"The
command ''{{command}}'' was cancelled because it took more than {{timeout}} seconds.
Please try a different command that completes more quickly."},"tools":{"filter":{"blocklist_error_template":"Operation
''{{action}}'' is not supported by this environment.","blocklist":["vim","vi","emacs","nano","nohup","gdb","less","tail
-f","python -m venv","make"],"blocklist_standalone":["python","python3","ipython","bash","sh","/bin/bash","/bin/sh","nohup","vi","vim","emacs","nano","su"],"block_unless_regex":{"radare2":"\\b(?:radare2)\\b.*\\s+-c\\s+.*","r2":"\\b(?:radare2)\\b.*\\s+-c\\s+.*"}},"bundles":[{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/registry","hidden_tools":[]},{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/edit_anthropic","hidden_tools":[]},{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/review_on_submit_m","hidden_tools":[]}],"env_variables":{},"registry_variables":{"USE_FILEMAP":"true","SUBMIT_REVIEW_MESSAGES":["Thank
you for your work on this issue. Please carefully follow the steps below to help
review your changes.\n\n1. If you made any changes to your code after running the
reproduction script, please run the reproduction script again.\n If the reproduction
script is failing, please revisit your changes and make sure they are correct.\n If
you have already removed your reproduction script, please ignore this step.\n2.
Remove your reproduction script (if you haven''t done so already).\n3. If you have
modified any TEST files, please revert them to the state they had before you started
fixing the issue.\n You can do this with `git checkout -- /path/to/test/file.py`.
Use below <diff> to find the files you need to revert.\n4. Run the submit command
again to confirm.\n\nHere is a list of all of your changes:\n\n<diff>\n{{diff}}\n</diff>\n"]},"submit_command":"submit","parse_function":{"error_message":"{%-
if error_code == \"missing\" -%}\nYour last output did not use any tool calls!\nPlease
make sure your output includes exactly _ONE_ function call!\nYou must invoke the
function directly using the function call format.\nYou cannot invoke commands with
```, you have to use the function call format.\nIf you think you have already resolved
the issue, please submit your changes by running the `submit` command.\nIf you think
you cannot solve the problem, please run `exit_forfeit` (if available) or `submit`.\nElse,
please continue with a new tool call!\n{%- elif error_code == \"multiple\" -%}\nYour
last output included multiple tool calls!\nPlease make sure your output includes
a thought and exactly _ONE_ function call.\n{%- elif error_code == \"unexpected_arg\"
-%}\nYour action could not be parsed properly: {{exception_message}}.\nMake sure
your function call doesn''t include any extra arguments that are not in the allowed
arguments, and only use the allowed commands.\n{%- else -%}\nYour action could not
be parsed properly: {{exception_message}}.\n{% endif %}\n","type":"function_calling"},"enable_bash_tool":true,"format_error_template":"{%-
if error_code == \"missing\" -%}\nYour last output did not use any tool calls!\nPlease
make sure your output includes exactly _ONE_ function call!\nYou must invoke the
function directly using the function call format.\nYou cannot invoke commands with
```, you have to use the function call format.\nIf you think you have already resolved
the issue, please submit your changes by running the `submit` command.\nIf you think
you cannot solve the problem, please run `exit_forfeit` (if available) or `submit`.\nElse,
please continue with a new tool call!\n{%- elif error_code == \"multiple\" -%}\nYour
last output included multiple tool calls!\nPlease make sure your output includes
a thought and exactly _ONE_ function call.\n{%- elif error_code == \"unexpected_arg\"
-%}\nYour action could not be parsed properly: {{exception_message}}.\nMake sure
your function call doesn''t include any extra arguments that are not in the allowed
arguments, and only use the allowed commands.\n{%- else -%}\nYour action could not
be parsed properly: {{exception_message}}.\n{% endif %}\n","command_docs":"bash:\n docstring:
runs the given command directly in bash\n signature: <command>\n arguments:\n -
command (string) [required]: The bash command to execute.\n\nstr_replace_editor:\n docstring:
Custom editing tool for viewing, creating and editing files * State is persistent
across command calls and discussions with the user * If `path` is a file, `view`
displays the result of applying `cat -n`. If `path` is a directory, `view` lists
non-hidden files and directories up to 2 levels deep * The `create` command cannot
be used if the specified `path` already exists as a file * If a `command` generates
a long output, it will be truncated and marked with `<response clipped>` * The `undo_edit`
command will revert the last edit made to the file at `path`\nNotes for using the
`str_replace` command: * The `old_str` parameter should match EXACTLY one or more
consecutive lines from the original file. Be mindful of whitespaces! * If the `old_str`
parameter is not unique in the file, the replacement will not be performed. Make
sure to include enough context in `old_str` to make it unique * The `new_str` parameter
should contain the edited lines that should replace the `old_str`\n\n signature:
str_replace_editor <command> <path> [<file_text>] [<view_range>] [<old_str>] [<new_str>]
[<insert_line>]\n\n arguments:\n - command (string) [required]: The commands
to run. Allowed options are: `view`, `create`, `str_replace`, `insert`, `undo_edit`.\n -
path (string) [required]: Absolute path to file or directory, e.g. `/testbed/file.py`
or `/testbed`.\n - file_text (string) [optional]: Required parameter of `create`
command, with the content of the file to be created.\n - old_str (string) [optional]:
Required parameter of `str_replace` command containing the string in `path` to replace.\n -
new_str (string) [optional]: Optional parameter of `str_replace` command containing
the new string (if not given, no string will be added). Required parameter of `insert`
command containing the string to insert.\n - insert_line (integer) [optional]:
Required parameter of `insert` command. The `new_str` will be inserted AFTER the
line `insert_line` of `path`.\n - view_range (array) [optional]: Optional parameter
of `view` command when `path` points to a file. If none is given, the full file
is shown. If provided, the file will be shown in the indicated line number range,
e.g. [11, 12] will show lines 11 and 12. Indexing at 1 to start. Setting `[start_line,
-1]` shows all lines from `start_line` to the end of the file.\n\nsubmit:\n docstring:
submits the current file\n signature: submit\n\n","multi_line_command_endings":{},"submit_command_end_name":null,"reset_commands":[],"execution_timeout":30,"install_timeout":300,"total_execution_timeout":1800,"max_consecutive_execution_timeouts":3},"history_processors":[{"type":"cache_control","last_n_messages":2,"last_n_messages_offset":0,"tagged_roles":["user","tool"]}],"model":{"name":"claude-3-sonnet-20240229","per_instance_cost_limit":3.0,"total_cost_limit":0.0,"per_instance_call_limit":0,"temperature":0.0,"top_p":1.0,"api_base":null,"api_version":null,"api_key":null,"stop":[],"completion_kwargs":{},"convert_system_to_user":false,"retry":{"retries":20,"min_wait":10.0,"max_wait":120.0},"delay":0.0,"fallbacks":[],"choose_api_key_by_thread":true,"max_input_tokens":null,"max_output_tokens":null},"max_requeries":3,"action_sampler":null,"type":"default"},"problem_statement":{"path":"debug_tests.md","extra_fields":{},"type":"text_file","id":"db62b9"},"output_dir":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/trajectories/tosinakinosho/anthropic_filemap__claude-3-sonnet-20240229__t-0.00__p-1.00__c-3.00___db62b9","actions":{"open_pr":false,"pr_config":{"skip_if_commits_reference_issue":true},"apply_patch_locally":false},"env_var_path":null}'
```
--------------------------------------------------------------------------------
/scripts/compile_requirements.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
# This script compiles requirements.in to requirements.txt using pip-compile
# Following the project's build standards for reproducible environments
set -e
# Default Python version if not specified
DEFAULT_VERSION="3.11"
PYTHON_VERSION=${1:-$DEFAULT_VERSION}
# Validate Python version
if [[ ! "$PYTHON_VERSION" =~ ^3\.(10|11|12|13)$ ]]; then
echo "Error: Python version must be 3.10, 3.11, 3.12 or 3.13."
echo "Usage: $0 [python-version]"
echo "Example: $0 3.10"
exit 1
fi
# Set the virtual environment directory based on the Python version
VENV_DIR=".compile-venv-py$PYTHON_VERSION"
# Check for private repository configuration
PRIVATE_REPO_URL=${PRIVATE_REPO_URL:-""}
PRIVATE_REPO_TOKEN=${PRIVATE_REPO_TOKEN:-""}
# Check for local package paths (comma-separated list of directories)
LOCAL_PACKAGE_PATHS=${LOCAL_PACKAGE_PATHS:-""}
echo "=========================================================="
echo "Compiling requirements for Python $PYTHON_VERSION"
echo "=========================================================="
# Create a Python virtual environment if it doesn't exist
if [ ! -d "$VENV_DIR" ]; then
echo "Creating a Python $PYTHON_VERSION virtual environment in $VENV_DIR..."
# Try different ways to create the environment based on the version
if command -v "python$PYTHON_VERSION" &> /dev/null; then
"python$PYTHON_VERSION" -m venv "$VENV_DIR"
elif command -v "python3.$PYTHON_VERSION" &> /dev/null; then
"python3.$PYTHON_VERSION" -m venv "$VENV_DIR"
else
echo "Error: Python $PYTHON_VERSION is not installed."
echo "Please install it and try again."
exit 1
fi
fi
# Activate the virtual environment
source "$VENV_DIR/bin/activate"
echo "Activated virtual environment: $VENV_DIR"
# Update pip and setuptools
echo "Updating pip and setuptools..."
pip install --upgrade pip setuptools wheel
# Install pip-tools
echo "Installing pip-tools..."
pip install pip-tools
# Make a backup of current requirements.txt if it exists
if [ -f "requirements-$PYTHON_VERSION.txt" ]; then
cp "requirements-$PYTHON_VERSION.txt" "requirements-$PYTHON_VERSION.txt.backup"
echo "Backed up existing requirements-$PYTHON_VERSION.txt to requirements-$PYTHON_VERSION.txt.backup"
fi
# Create a temporary copy of requirements.in with adjusted version constraints
cp requirements.in requirements.in.tmp
# Create pip.conf for private repository access if provided
if [ ! -z "$PRIVATE_REPO_URL" ]; then
mkdir -p "$VENV_DIR/pip"
cat > "$VENV_DIR/pip/pip.conf" << EOF
[global]
index-url = https://pypi.org/simple
extra-index-url = ${PRIVATE_REPO_URL}
EOF
if [ ! -z "$PRIVATE_REPO_TOKEN" ]; then
echo "Using private repository with authentication token"
# Add credentials to pip.conf if token is provided
sed -i.bak "s|${PRIVATE_REPO_URL}|${PRIVATE_REPO_URL}:${PRIVATE_REPO_TOKEN}@|" "$VENV_DIR/pip/pip.conf" 2>/dev/null || \
sed -i '' "s|${PRIVATE_REPO_URL}|${PRIVATE_REPO_URL}:${PRIVATE_REPO_TOKEN}@|" "$VENV_DIR/pip/pip.conf"
fi
export PIP_CONFIG_FILE="$VENV_DIR/pip/pip.conf"
fi
# Parse and set up local package paths if provided
LOCAL_ARGS=""
if [ ! -z "$LOCAL_PACKAGE_PATHS" ]; then
echo "Setting up local package paths..."
IFS=',' read -ra PATHS <<< "$LOCAL_PACKAGE_PATHS"
for path in "${PATHS[@]}"; do
LOCAL_ARGS="$LOCAL_ARGS -f $path"
done
echo "Local package paths: $LOCAL_ARGS"
fi
# Check for local git repositories
if [ -d "./local-packages" ]; then
echo "Found local-packages directory, will include in search path"
LOCAL_ARGS="$LOCAL_ARGS -f ./local-packages"
fi
# Fix for dependency issues - version-specific adjustments
echo "Adjusting dependency constraints for compatibility with Python $PYTHON_VERSION..."
# Version-specific adjustments
if [ "$PYTHON_VERSION" = "3.9" ]; then
# Python 3.9-specific adjustments
sed -i.bak 's/torch>=2.0.0/torch>=1.13.0,<2.0.0/' requirements.in.tmp 2>/dev/null || sed -i '' 's/torch>=2.0.0/torch>=1.13.0,<2.0.0/' requirements.in.tmp
sed -i.bak 's/networkx>=.*$/networkx>=2.8.0,<3.0/' requirements.in.tmp 2>/dev/null || sed -i '' 's/networkx>=.*$/networkx>=2.8.0,<3.0/' requirements.in.tmp
# Keep starlette constraint for Python 3.9
elif [ "$PYTHON_VERSION" = "3.10" ] || [ "$PYTHON_VERSION" = "3.11" ] || [ "$PYTHON_VERSION" = "3.12" ] || [ "$PYTHON_VERSION" = "3.13" ]; then
# Python 3.10/3.11-specific adjustments
sed -i.bak 's/networkx>=.*$/networkx>=2.8.0/' requirements.in.tmp 2>/dev/null || sed -i '' 's/networkx>=.*$/networkx>=2.8.0/' requirements.in.tmp
# Modify starlette constraint for Python 3.10/3.11 (for diagnostic purposes)
# Also apply for Python 3.12/3.13
echo "Modifying starlette constraint for Python $PYTHON_VERSION to diagnose dependency conflicts..."
sed -i.bak 's/starlette>=0.27.0,<0.28.0/starlette>=0.27.0/' requirements.in.tmp 2>/dev/null || \
sed -i '' 's/starlette>=0.27.0,<0.28.0/starlette>=0.27.0/' requirements.in.tmp
fi
# Special handling for private packages
COMPILE_SUCCESS=0
# Try to compile with all packages
echo "Compiling adjusted requirements.in to requirements-$PYTHON_VERSION.txt..."
if pip-compile --allow-unsafe $LOCAL_ARGS --output-file="requirements-$PYTHON_VERSION.txt" requirements.in.tmp; then
COMPILE_SUCCESS=1
echo "Compilation successful with all packages included."
else
echo "First compilation attempt failed, trying without private packages..."
fi
# If compilation with all packages failed, try without problematic private packages
if [ $COMPILE_SUCCESS -eq 0 ]; then
echo "Creating a version without private packages..."
grep -v "uvx\|mcp-server-qdrant" requirements.in > requirements.in.basic
# Add version-specific constraints
if [ "$PYTHON_VERSION" = "3.9" ]; then
echo "# Conservative dependencies for Python 3.9" >> requirements.in.basic
echo "networkx>=2.8.0,<3.0" >> requirements.in.basic
echo "torch>=1.13.0,<2.0.0" >> requirements.in.basic
# Keep original starlette constraint
grep "starlette" requirements.in >> requirements.in.basic
elif [ "$PYTHON_VERSION" = "3.10" ] || [ "$PYTHON_VERSION" = "3.11" ] || [ "$PYTHON_VERSION" = "3.12" ] || [ "$PYTHON_VERSION" = "3.13" ]; then
echo "# Conservative dependencies for Python $PYTHON_VERSION" >> requirements.in.basic
echo "networkx>=2.8.0" >> requirements.in.basic
# Modified starlette constraint for 3.10/3.11
echo "starlette>=0.27.0" >> requirements.in.basic
fi
if pip-compile --allow-unsafe $LOCAL_ARGS --output-file="requirements-$PYTHON_VERSION.txt" requirements.in.basic; then
COMPILE_SUCCESS=1
echo "Compilation successful without private packages."
echo "# NOTE: Private packages (uvx, mcp-server-qdrant) were excluded from this compilation." >> "requirements-$PYTHON_VERSION.txt"
echo "# You may need to install them separately from their source." >> "requirements-$PYTHON_VERSION.txt"
# Create a separate file just for private packages
echo "# Private packages excluded from main requirements-$PYTHON_VERSION.txt" > "requirements-private-$PYTHON_VERSION.txt"
grep "uvx\|mcp-server-qdrant" requirements.in >> "requirements-private-$PYTHON_VERSION.txt"
echo "Created separate requirements-private-$PYTHON_VERSION.txt for private packages."
else
echo "WARNING: Both compilation attempts failed. Please check for compatibility issues."
# Additional diagnostic information
echo "Failed compilation error log:"
if [ "$PYTHON_VERSION" = "3.10" ] || [ "$PYTHON_VERSION" = "3.11" ]; then
echo "Testing if removing starlette constraint entirely resolves the issue..."
grep -v "starlette\|uvx\|mcp-server-qdrant" requirements.in > requirements.in.minimal
echo "# Minimal dependencies for Python $PYTHON_VERSION" >> requirements.in.minimal
echo "networkx>=2.8.0" >> requirements.in.minimal
if pip-compile --allow-unsafe $LOCAL_ARGS --output-file="requirements-$PYTHON_VERSION.minimal.txt" requirements.in.minimal; then
echo "SUCCESS: Compilation successful without starlette constraint."
echo "This confirms that starlette is causing dependency conflicts."
# Create a working requirements file for now
mv "requirements-$PYTHON_VERSION.minimal.txt" "requirements-$PYTHON_VERSION.txt"
echo "# WARNING: starlette constraint was removed to resolve conflicts" >> "requirements-$PYTHON_VERSION.txt"
echo "# You will need to manually install a compatible starlette version" >> "requirements-$PYTHON_VERSION.txt"
COMPILE_SUCCESS=1
else
echo "FAILURE: Issue persists even without starlette constraint."
fi
fi
fi
fi
# Create a symlink or copy of the default version to requirements.txt
if [ "$PYTHON_VERSION" = "$DEFAULT_VERSION" ]; then
echo "Creating requirements.txt as copy of requirements-$PYTHON_VERSION.txt (default version)"
cp "requirements-$PYTHON_VERSION.txt" requirements.txt
# Also copy private requirements if they exist
if [ -f "requirements-private-$PYTHON_VERSION.txt" ]; then
cp "requirements-private-$PYTHON_VERSION.txt" requirements-private.txt
fi
fi
# Clean up temporary files
rm -f requirements.in.tmp requirements.in.tmp.bak requirements.in.bak requirements.in.basic requirements.in.minimal 2>/dev/null || true
# Show generated file
echo "Compilation complete. Generated requirements-$PYTHON_VERSION.txt with pinned dependencies."
echo ""
echo "To use private package repositories, set environment variables before running this script:"
echo " export PRIVATE_REPO_URL=\"https://your-private-repo.com/simple\""
echo " export PRIVATE_REPO_TOKEN=\"your-access-token\" # Optional"
echo ""
echo "To use local package paths, set LOCAL_PACKAGE_PATHS:"
echo " export LOCAL_PACKAGE_PATHS=\"/path/to/packages1,/path/to/packages2\""
echo ""
echo "You can specify a Python version when running this script:"
echo " ./scripts/compile_requirements.sh 3.9 # For Python 3.9"
echo " ./scripts/compile_requirements.sh 3.10 # For Python 3.10"
echo " ./scripts/compile_requirements.sh 3.11 # For Python 3.11"
# Optional: show differences if the file existed before
if [ -f "requirements-$PYTHON_VERSION.txt.backup" ]; then
echo "Changes from previous requirements-$PYTHON_VERSION.txt:"
diff -u "requirements-$PYTHON_VERSION.txt.backup" "requirements-$PYTHON_VERSION.txt" || true
fi
# Deactivate the virtual environment
deactivate
echo "Completed and deactivated virtual environment."
# Clean up the temporary venv if desired
read -p "Remove temporary virtual environment? (y/n) " -n 1 -r
echo
if [[ $REPLY =~ ^[Yy]$ ]]; then
rm -rf "$VENV_DIR"
echo "Removed temporary virtual environment."
fi
echo "Done."
```
--------------------------------------------------------------------------------
/src/mcp_codebase_insight/core/documentation.py:
--------------------------------------------------------------------------------
```python
"""Documentation management module."""
import json
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Dict, List, Optional
from uuid import UUID, uuid4
from urllib.parse import urlparse
from pydantic import BaseModel
class DocumentationType(str, Enum):
"""Documentation type enumeration."""
REFERENCE = "reference"
TUTORIAL = "tutorial"
API = "api"
GUIDE = "guide"
EXAMPLE = "example"
PATTERN = "pattern"
class Document(BaseModel):
"""Document model."""
id: UUID
title: str
type: DocumentationType
content: str
metadata: Optional[Dict[str, str]] = None
tags: Optional[List[str]] = None
created_at: datetime
updated_at: datetime
version: Optional[str] = None
related_docs: Optional[List[UUID]] = None
class DocumentationManager:
"""Manager for documentation handling."""
def __init__(self, config):
"""Initialize documentation manager."""
self.config = config
self.docs_dir = config.docs_cache_dir
self.docs_dir.mkdir(parents=True, exist_ok=True)
self.initialized = False
self.documents: Dict[UUID, Document] = {}
async def initialize(self):
"""Initialize the documentation manager.
This method ensures the docs directory exists and loads any existing documents.
"""
if self.initialized:
return
try:
# Ensure docs directory exists
self.docs_dir.mkdir(parents=True, exist_ok=True)
# Load any existing documents
for doc_file in self.docs_dir.glob("*.json"):
if doc_file.is_file():
try:
with open(doc_file, "r") as f:
doc_data = json.load(f)
# Convert the loaded data into a Document object
doc = Document(**doc_data)
self.documents[doc.id] = doc
except (json.JSONDecodeError, ValueError) as e:
# Log error but continue processing other files
print(f"Error loading document {doc_file}: {e}")
self.initialized = True
except Exception as e:
print(f"Error initializing documentation manager: {e}")
await self.cleanup()
raise RuntimeError(f"Failed to initialize documentation manager: {str(e)}")
async def cleanup(self):
"""Clean up resources used by the documentation manager.
This method ensures all documents are saved and resources are released.
"""
if not self.initialized:
return
try:
# Save any modified documents
for doc in self.documents.values():
try:
await self._save_document(doc)
except Exception as e:
print(f"Error saving document {doc.id}: {e}")
# Clear in-memory documents
self.documents.clear()
except Exception as e:
print(f"Error cleaning up documentation manager: {e}")
finally:
self.initialized = False
async def add_document(
self,
title: str,
content: str,
type: DocumentationType,
metadata: Optional[Dict[str, str]] = None,
tags: Optional[List[str]] = None,
version: Optional[str] = None,
related_docs: Optional[List[UUID]] = None
) -> Document:
"""Add a new document."""
now = datetime.utcnow()
doc = Document(
id=uuid4(),
title=title,
type=type,
content=content,
metadata=metadata,
tags=tags,
version=version,
related_docs=related_docs,
created_at=now,
updated_at=now
)
await self._save_document(doc)
return doc
async def get_document(self, doc_id: UUID) -> Optional[Document]:
"""Get document by ID."""
doc_path = self.docs_dir / f"{doc_id}.json"
if not doc_path.exists():
return None
with open(doc_path) as f:
data = json.load(f)
return Document(**data)
async def update_document(
self,
doc_id: UUID,
content: Optional[str] = None,
metadata: Optional[Dict[str, str]] = None,
tags: Optional[List[str]] = None,
version: Optional[str] = None,
related_docs: Optional[List[UUID]] = None
) -> Optional[Document]:
"""Update document content and metadata."""
doc = await self.get_document(doc_id)
if not doc:
return None
if content:
doc.content = content
if metadata:
doc.metadata = {**(doc.metadata or {}), **metadata}
if tags:
doc.tags = tags
if version:
doc.version = version
if related_docs:
doc.related_docs = related_docs
doc.updated_at = datetime.utcnow()
await self._save_document(doc)
return doc
async def list_documents(
self,
type: Optional[DocumentationType] = None,
tags: Optional[List[str]] = None
) -> List[Document]:
"""List all documents, optionally filtered by type and tags."""
docs = []
for path in self.docs_dir.glob("*.json"):
with open(path) as f:
data = json.load(f)
doc = Document(**data)
# Apply filters
if type and doc.type != type:
continue
if tags and not all(tag in (doc.tags or []) for tag in tags):
continue
docs.append(doc)
return sorted(docs, key=lambda x: x.created_at)
async def search_documents(
self,
query: str,
type: Optional[DocumentationType] = None,
tags: Optional[List[str]] = None,
limit: int = 10
) -> List[Document]:
"""Search documents by content."""
# TODO: Implement proper text search
# For now, just do simple substring matching
results = []
query = query.lower()
for doc in await self.list_documents(type, tags):
if (
query in doc.title.lower() or
query in doc.content.lower() or
any(query in tag.lower() for tag in (doc.tags or []))
):
results.append(doc)
if len(results) >= limit:
break
return results
async def _save_document(self, doc: Document) -> None:
"""Save document to file."""
doc_path = self.docs_dir / f"{doc.id}.json"
with open(doc_path, "w") as f:
json.dump(doc.model_dump(), f, indent=2, default=str)
async def crawl_docs(
self,
urls: List[str],
source_type: str
) -> List[Document]:
"""Crawl documentation from URLs."""
import aiohttp
from bs4 import BeautifulSoup
docs = []
try:
doc_type = DocumentationType(source_type)
except ValueError:
doc_type = DocumentationType.REFERENCE
async with aiohttp.ClientSession() as session:
for url in urls:
try:
# Handle file URLs specially (for testing)
parsed_url = urlparse(url)
if parsed_url.scheme == "file":
# Create a test document
doc = await self.add_document(
title="Test Documentation",
content="This is a test document for testing the documentation crawler.",
type=doc_type,
metadata={
"source_url": url,
"source_type": source_type,
"crawled_at": datetime.utcnow().isoformat()
}
)
docs.append(doc)
continue
# Fetch the content
async with session.get(url, timeout=10) as response:
if response.status != 200:
print(f"Error fetching {url}: HTTP {response.status}")
continue
content = await response.text()
# Parse HTML content
soup = BeautifulSoup(content, 'html.parser')
# Extract title from meta tags or h1
title = soup.find('meta', property='og:title')
if title:
title = title.get('content')
else:
title = soup.find('h1')
if title:
title = title.text.strip()
else:
title = f"Documentation from {url}"
# Extract main content
# First try to find main content area
content = ""
main = soup.find('main')
if main:
content = main.get_text(separator='\n', strip=True)
else:
# Try article tag
article = soup.find('article')
if article:
content = article.get_text(separator='\n', strip=True)
else:
# Fallback to body content
body = soup.find('body')
if body:
content = body.get_text(separator='\n', strip=True)
else:
content = soup.get_text(separator='\n', strip=True)
# Create document
doc = await self.add_document(
title=title,
content=content,
type=doc_type,
metadata={
"source_url": url,
"source_type": source_type,
"crawled_at": datetime.utcnow().isoformat()
}
)
docs.append(doc)
except Exception as e:
# Log error but continue with other URLs
print(f"Error crawling {url}: {str(e)}")
continue
return docs
```
--------------------------------------------------------------------------------
/tests/integration/test_communication_integration.py:
--------------------------------------------------------------------------------
```python
import asyncio
import json
import pytest
from unittest.mock import MagicMock, AsyncMock
from tests.components.test_stdio_components import MockStdinReader, MockStdoutWriter
class MockSSEClient:
def __init__(self):
self.events = []
self.connected = True
async def send(self, event):
if not self.connected:
raise ConnectionError("Client disconnected")
self.events.append(event)
def disconnect(self):
self.connected = False
@pytest.fixture
async def mock_communication_setup():
"""Set up mock stdio and SSE components for integration testing."""
# Set up stdio mocks
stdio_reader = MockStdinReader("")
stdio_writer = MockStdoutWriter()
# Set up SSE mock
sse_client = MockSSEClient()
return stdio_reader, stdio_writer, sse_client
@pytest.mark.asyncio
async def test_sse_stdio_interaction(mock_communication_setup):
"""Test interaction between SSE and STDIO communication channels."""
stdio_reader, stdio_writer, sse_client = await mock_communication_setup
# Step 1: Tool registration via STDIO
registration_message = {
"type": "register",
"tool_id": "test_tool",
"capabilities": ["capability1", "capability2"]
}
# Override reader's input with registration message
stdio_reader.input_stream.write(json.dumps(registration_message) + "\n")
stdio_reader.input_stream.seek(0)
# Process registration
line = await stdio_reader.readline()
message = json.loads(line)
# Send registration acknowledgment via stdio
response = {
"type": "registration_success",
"tool_id": message["tool_id"]
}
await stdio_writer.write(json.dumps(response) + "\n")
# Send SSE notification about new tool
sse_notification = {
"type": "tool_registered",
"tool_id": message["tool_id"],
"capabilities": message["capabilities"]
}
await sse_client.send(json.dumps(sse_notification))
# Verify stdio response
assert "registration_success" in stdio_writer.get_output()
# Verify SSE notification
assert len(sse_client.events) == 1
assert "tool_registered" in sse_client.events[0]
assert message["tool_id"] in sse_client.events[0]
# Step 2: SSE event triggering STDIO message
# Reset the writer to clear previous output
stdio_writer = MockStdoutWriter()
# Simulate an SSE event that should trigger a STDIO message
sse_event = {
"type": "request",
"id": "sse_to_stdio_test",
"method": "test_method",
"params": {"param1": "value1"}
}
# In a real system, this would be processed by an event handler
# that would then write to STDIO. Here we simulate that directly.
await sse_client.send(json.dumps(sse_event))
# Simulate the STDIO response that would be generated
stdio_response = {
"type": "response",
"id": sse_event["id"],
"result": {"status": "success"}
}
await stdio_writer.write(json.dumps(stdio_response) + "\n")
# Verify the STDIO response
assert "response" in stdio_writer.get_output()
assert sse_event["id"] in stdio_writer.get_output()
# Step 3: Bidirectional communication with state tracking
# Create a simple state tracker
state = {"last_message_id": None, "message_count": 0}
# Send a sequence of messages in both directions
for i in range(3):
# STDIO to SSE
stdio_message = {
"type": "notification",
"id": f"msg_{i}",
"data": f"data_{i}"
}
# In a real system, this would come from STDIO input
# Here we simulate by updating state directly
state["last_message_id"] = stdio_message["id"]
state["message_count"] += 1
# Send to SSE
await sse_client.send(json.dumps(stdio_message))
# SSE to STDIO
sse_response = {
"type": "event",
"id": f"response_{i}",
"in_response_to": stdio_message["id"],
"data": f"response_data_{i}"
}
# Process SSE response and update STDIO
await stdio_writer.write(json.dumps(sse_response) + "\n")
# Verify the communication flow
assert state["message_count"] == 3
assert state["last_message_id"] == "msg_2"
assert len(sse_client.events) == 5 # 1 from registration + 1 from SSE event + 3 from the loop
# Verify STDIO output contains all responses
stdio_output = stdio_writer.get_output()
for i in range(3):
assert f"response_{i}" in stdio_output
assert f"response_data_{i}" in stdio_output
@pytest.mark.asyncio
async def test_bidirectional_communication(mock_communication_setup):
"""Test bidirectional communication between stdio and SSE."""
stdio_reader, stdio_writer, sse_client = await mock_communication_setup
# Set up test message flow
stdio_messages = [
{"type": "request", "id": "1", "method": "test", "data": "stdio_data"},
{"type": "request", "id": "2", "method": "test", "data": "more_data"}
]
# Write messages to stdio
for msg in stdio_messages:
stdio_reader.input_stream.write(json.dumps(msg) + "\n")
stdio_reader.input_stream.seek(0)
# Process messages and generate SSE events
while True:
line = await stdio_reader.readline()
if not line:
break
# Process stdio message
message = json.loads(line)
# Generate SSE event
sse_event = {
"type": "event",
"source": "stdio",
"data": message["data"]
}
await sse_client.send(json.dumps(sse_event))
# Send response via stdio
response = {
"type": "response",
"id": message["id"],
"status": "success"
}
await stdio_writer.write(json.dumps(response) + "\n")
# Verify all messages were processed
assert len(sse_client.events) == len(stdio_messages)
assert all("stdio" in event for event in sse_client.events)
# Verify stdio responses
output = stdio_writer.get_output()
responses = [json.loads(line) for line in output.strip().split("\n")]
assert len(responses) == len(stdio_messages)
assert all(resp["type"] == "response" for resp in responses)
@pytest.mark.asyncio
async def test_error_propagation(mock_communication_setup):
"""Test error propagation between stdio and SSE."""
stdio_reader, stdio_writer, sse_client = await mock_communication_setup
# Simulate error in stdio
error_message = {
"type": "request",
"id": "error_test",
"method": "test",
"data": "error_data"
}
stdio_reader.input_stream.write(json.dumps(error_message) + "\n")
stdio_reader.input_stream.seek(0)
# Process message and simulate error
line = await stdio_reader.readline()
message = json.loads(line)
# Generate error response in stdio
error_response = {
"type": "error",
"id": message["id"],
"error": "Test error occurred"
}
await stdio_writer.write(json.dumps(error_response) + "\n")
# Propagate error to SSE
sse_error_event = {
"type": "error_event",
"source": "stdio",
"error": "Test error occurred",
"request_id": message["id"]
}
await sse_client.send(json.dumps(sse_error_event))
# Verify error handling
assert "error" in stdio_writer.get_output()
assert len(sse_client.events) == 1
assert "error_event" in sse_client.events[0]
@pytest.mark.asyncio
async def test_connection_state_handling(mock_communication_setup):
"""Test handling of connection state changes."""
stdio_reader, stdio_writer, sse_client = await mock_communication_setup
# Test normal operation
test_message = {
"type": "request",
"id": "state_test",
"method": "test"
}
stdio_reader.input_stream.write(json.dumps(test_message) + "\n")
stdio_reader.input_stream.seek(0)
# Process message while connected
line = await stdio_reader.readline()
message = json.loads(line)
await sse_client.send(json.dumps({"type": "event", "data": "test"}))
# Simulate SSE client disconnect
sse_client.disconnect()
# Attempt to send message after disconnect
with pytest.raises(ConnectionError):
await sse_client.send(json.dumps({"type": "event", "data": "test"}))
# Send disconnect notification via stdio
disconnect_notification = {
"type": "notification",
"event": "client_disconnected"
}
await stdio_writer.write(json.dumps(disconnect_notification) + "\n")
# Verify disconnect handling
assert "client_disconnected" in stdio_writer.get_output()
assert not sse_client.connected
@pytest.mark.asyncio
async def test_race_condition_handling(mock_communication_setup):
"""Test handling of potential race conditions in message processing."""
stdio_reader, stdio_writer, sse_client = await mock_communication_setup
messages = [
{"type": "request", "id": f"race_test_{i}", "sequence": i, "data": f"data_{i}"}
for i in range(5)
]
import random
shuffled_messages = messages.copy()
random.shuffle(shuffled_messages)
for msg in shuffled_messages:
stdio_reader.input_stream.write(json.dumps(msg) + "\n")
stdio_reader.input_stream.seek(0)
received_messages = {}
while True:
line = await stdio_reader.readline()
if not line:
break
message = json.loads(line)
received_messages[message["sequence"]] = message
await sse_client.send(json.dumps({
"type": "event",
"sequence": message["sequence"],
"data": message["data"]
}))
await stdio_writer.write(json.dumps({
"type": "response",
"id": message["id"],
"sequence": message["sequence"]
}) + "\n")
ordered_sequences = sorted(received_messages.keys())
assert ordered_sequences == list(range(5))
for i, event_json in enumerate(sse_client.events):
event = json.loads(event_json)
assert event["sequence"] < len(messages)
@pytest.mark.asyncio
async def test_resource_cleanup(mock_communication_setup):
"""Test proper cleanup of resources after communication ends."""
stdio_reader, stdio_writer, sse_client = await mock_communication_setup
allocated_resources = set()
async def allocate_resource(resource_id):
allocated_resources.add(resource_id)
async def release_resource(resource_id):
allocated_resources.remove(resource_id)
message = {"type": "request", "id": "resource_test", "resource": "test_resource"}
stdio_reader.input_stream.write(json.dumps(message) + "\n")
stdio_reader.input_stream.seek(0)
line = await stdio_reader.readline()
message = json.loads(line)
resource_id = message["resource"]
await allocate_resource(resource_id)
try:
await asyncio.sleep(0.1)
await stdio_writer.write(json.dumps({
"type": "response",
"id": message["id"],
"status": "success"
}) + "\n")
finally:
await release_resource(resource_id)
assert len(allocated_resources) == 0
@pytest.mark.asyncio
async def test_partial_message_handling(mock_communication_setup):
"""Test handling of partial or truncated messages."""
stdio_reader, stdio_writer, sse_client = await mock_communication_setup
partial_json = '{"type": "request", "id": "partial_test", "method": "test"'
stdio_reader.input_stream.write(partial_json + "\n")
stdio_reader.input_stream.seek(0)
line = await stdio_reader.readline()
try:
json.loads(line)
parsed = True
except json.JSONDecodeError:
parsed = False
error_response = {
"type": "error",
"error": "Invalid JSON format",
"code": "PARSE_ERROR"
}
await stdio_writer.write(json.dumps(error_response) + "\n")
assert not parsed, "Parsing should have failed with partial JSON"
assert "Invalid JSON format" in stdio_writer.get_output()
assert "PARSE_ERROR" in stdio_writer.get_output()
```
--------------------------------------------------------------------------------
/scripts/load_example_patterns.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""Load example patterns and ADRs into the knowledge base."""
import asyncio
import json
from pathlib import Path
from datetime import datetime
from uuid import uuid4
from mcp_codebase_insight.core.config import ServerConfig
from mcp_codebase_insight.core.knowledge import KnowledgeBase, Pattern, PatternType, PatternConfidence
from mcp_codebase_insight.core.vector_store import VectorStore
from mcp_codebase_insight.core.embeddings import SentenceTransformerEmbedding
from mcp_codebase_insight.core.adr import ADRManager, ADRStatus
# Example patterns data
PATTERNS = [
{
"name": "Factory Method",
"type": "design_pattern",
"description": "Define an interface for creating an object, but let subclasses decide which class to instantiate.",
"content": """
class Creator:
def factory_method(self):
pass
def operation(self):
product = self.factory_method()
return product.operation()
class ConcreteCreator(Creator):
def factory_method(self):
return ConcreteProduct()
""",
"tags": ["creational", "factory", "object-creation"],
"confidence": "high"
},
{
"name": "Repository Pattern",
"type": "architecture",
"description": "Mediates between the domain and data mapping layers using a collection-like interface for accessing domain objects.",
"content": """
class Repository:
def get(self, id: str) -> Entity:
pass
def add(self, entity: Entity):
pass
def remove(self, entity: Entity):
pass
""",
"tags": ["data-access", "persistence", "domain-driven-design"],
"confidence": "high"
},
{
"name": "Strategy Pattern",
"type": "design_pattern",
"description": "Define a family of algorithms, encapsulate each one, and make them interchangeable.",
"content": """
class Strategy:
def execute(self, data):
pass
class ConcreteStrategyA(Strategy):
def execute(self, data):
return "Algorithm A"
class Context:
def __init__(self, strategy: Strategy):
self._strategy = strategy
def execute_strategy(self, data):
return self._strategy.execute(data)
""",
"tags": ["behavioral", "algorithm", "encapsulation"],
"confidence": "high"
},
{
"name": "Error Handling Pattern",
"type": "code",
"description": "Common pattern for handling errors in Python using try-except with context.",
"content": """
def operation_with_context():
try:
# Setup resources
resource = setup_resource()
try:
# Main operation
result = process_resource(resource)
return result
except SpecificError as e:
# Handle specific error
handle_specific_error(e)
raise
finally:
# Cleanup
cleanup_resource(resource)
except Exception as e:
# Log error with context
logger.error("Operation failed", exc_info=e)
raise OperationError("Operation failed") from e
""",
"tags": ["error-handling", "python", "best-practice"],
"confidence": "high"
},
{
"name": "Circuit Breaker",
"type": "architecture",
"description": "Prevent system failure by failing fast and handling recovery.",
"content": """
class CircuitBreaker:
def __init__(self, failure_threshold, reset_timeout):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self.last_failure_time = None
self.state = "closed"
async def call(self, func, *args, **kwargs):
if self._should_open():
self.state = "open"
raise CircuitBreakerOpen()
try:
result = await func(*args, **kwargs)
self._reset()
return result
except Exception as e:
self._record_failure()
raise
""",
"tags": ["resilience", "fault-tolerance", "microservices"],
"confidence": "high"
}
]
# Example ADRs data
ADRS = [
{
"title": "Use FastAPI for REST API Development",
"context": {
"problem": "We need a modern, high-performance web framework for our REST API",
"constraints": [
"Must support Python 3.9+",
"Must support async/await",
"Must have strong type validation",
"Must have good documentation"
],
"assumptions": [
"The team has Python experience",
"Performance is a priority"
]
},
"options": [
{
"title": "Use Flask",
"pros": [
"Simple and familiar",
"Large ecosystem",
"Easy to learn"
],
"cons": [
"No built-in async support",
"No built-in validation",
"Requires many extensions"
]
},
{
"title": "Use FastAPI",
"pros": [
"Built-in async support",
"Automatic OpenAPI documentation",
"Built-in validation with Pydantic",
"High performance"
],
"cons": [
"Newer framework with smaller ecosystem",
"Steeper learning curve for some concepts"
]
},
{
"title": "Use Django REST Framework",
"pros": [
"Mature and stable",
"Full-featured",
"Large community"
],
"cons": [
"Heavier weight",
"Limited async support",
"Slower than alternatives"
]
}
],
"decision": "We will use FastAPI for our REST API development due to its modern features, performance, and built-in support for async/await and validation.",
"consequences": {
"positive": [
"Improved API performance",
"Better developer experience with type hints and validation",
"Automatic API documentation"
],
"negative": [
"Team needs to learn new concepts (dependency injection, Pydantic)",
"Fewer third-party extensions compared to Flask or Django"
]
}
},
{
"title": "Vector Database for Semantic Search",
"context": {
"problem": "We need a database solution for storing and searching vector embeddings for semantic code search",
"constraints": [
"Must support efficient vector similarity search",
"Must scale to handle large codebases",
"Must be easy to integrate with Python"
]
},
"options": [
{
"title": "Use Qdrant",
"pros": [
"Purpose-built for vector search",
"Good Python client",
"Fast similarity search",
"Support for filters"
],
"cons": [
"Relatively new project",
"Limited community compared to alternatives"
]
},
{
"title": "Use Elasticsearch with vector capabilities",
"pros": [
"Mature product",
"Well-known in industry",
"Many features beyond vector search"
],
"cons": [
"More complex to set up",
"Not optimized exclusively for vector search",
"Higher resource requirements"
]
},
{
"title": "Build custom solution with NumPy/FAISS",
"pros": [
"Complete control over implementation",
"No external service dependency",
"Can optimize for specific needs"
],
"cons": [
"Significant development effort",
"Need to handle persistence manually",
"Maintenance burden"
]
}
],
"decision": "We will use Qdrant for vector storage and similarity search due to its performance, ease of use, and purpose-built design for vector operations.",
"consequences": {
"positive": [
"Fast similarity search with minimal setup",
"Simple API for vector operations",
"Good scalability as codebase grows"
],
"negative": [
"New dependency to maintain",
"Team needs to learn Qdrant-specific concepts"
]
}
}
]
async def main():
"""Load patterns and ADRs into knowledge base."""
try:
# Create config
config = ServerConfig()
# Initialize components
embedder = SentenceTransformerEmbedding(config.embedding_model)
vector_store = VectorStore(
url=config.qdrant_url,
embedder=embedder,
collection_name=config.collection_name,
vector_name="fast-all-minilm-l6-v2"
)
# Initialize vector store
await vector_store.initialize()
# Create knowledge base
kb = KnowledgeBase(config, vector_store)
await kb.initialize()
# Create patterns directory if it doesn't exist
patterns_dir = Path("knowledge/patterns")
patterns_dir.mkdir(parents=True, exist_ok=True)
# Create ADRs directory if it doesn't exist
adrs_dir = Path("docs/adrs")
adrs_dir.mkdir(parents=True, exist_ok=True)
# Load each pattern
print("\n=== Loading Patterns ===")
for pattern_data in PATTERNS:
# Save pattern to knowledge base using the correct method signature
created = await kb.add_pattern(
name=pattern_data["name"],
type=PatternType(pattern_data["type"]),
description=pattern_data["description"],
content=pattern_data["content"],
confidence=PatternConfidence(pattern_data["confidence"]),
tags=pattern_data["tags"]
)
print(f"Added pattern: {created.name}")
# Save pattern to file
pattern_file = patterns_dir / f"{created.id}.json"
with open(pattern_file, "w") as f:
json.dump({
"id": str(created.id),
"name": created.name,
"type": created.type.value,
"description": created.description,
"content": created.content,
"tags": created.tags,
"confidence": created.confidence.value,
"created_at": created.created_at.isoformat(),
"updated_at": created.updated_at.isoformat()
}, f, indent=2)
print("\nAll patterns loaded successfully!")
# Initialize ADR manager
print("\n=== Loading ADRs ===")
adr_manager = ADRManager(config)
await adr_manager.initialize()
# Load each ADR
for adr_data in ADRS:
created = await adr_manager.create_adr(
title=adr_data["title"],
context=adr_data["context"],
options=adr_data["options"],
decision=adr_data["decision"],
consequences=adr_data.get("consequences")
)
print(f"Added ADR: {created.title}")
print("\nAll ADRs loaded successfully!")
# Test pattern search
print("\n=== Testing Pattern Search ===")
results = await kb.find_similar_patterns(
"error handling in Python",
limit=2
)
print("\nSearch results:")
for result in results:
print(f"- {result.pattern.name} (score: {result.similarity_score:.2f})")
# Test ADR listing
print("\n=== Testing ADR Listing ===")
adrs = await adr_manager.list_adrs()
print(f"\nFound {len(adrs)} ADRs:")
for adr in adrs:
print(f"- {adr.title} (status: {adr.status})")
except Exception as e:
print(f"Error loading examples: {e}")
raise
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/tests/config/test_config_and_env.py:
--------------------------------------------------------------------------------
```python
"""Tests for configuration and environment handling."""
import sys
import os
# Ensure the src directory is in the Python path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))
import os
import asyncio
import shutil
import pytest
import pytest_asyncio
from pathlib import Path
from typing import Generator
from unittest.mock import patch
import uuid
from qdrant_client import QdrantClient
from qdrant_client.http.models import Distance, VectorParams
from src.mcp_codebase_insight.core.config import ServerConfig
from src.mcp_codebase_insight.server import CodebaseAnalysisServer
@pytest.fixture(scope="session")
def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
"""Create event loop for tests."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture
def env_vars(tmp_path):
"""Set up test environment variables and clean up test directories."""
original_env = dict(os.environ)
test_dirs = {
"MCP_DOCS_CACHE_DIR": tmp_path / "test_docs",
"MCP_ADR_DIR": tmp_path / "test_docs/adrs",
"MCP_KB_STORAGE_DIR": tmp_path / "test_knowledge",
"MCP_DISK_CACHE_DIR": tmp_path / "test_cache"
}
test_vars = {
"MCP_HOST": "127.0.0.1",
"MCP_PORT": "8000",
"MCP_LOG_LEVEL": "DEBUG",
"MCP_DEBUG": "true",
"MCP_METRICS_ENABLED": "true",
"MCP_CACHE_ENABLED": "true",
"MCP_QDRANT_URL": "http://localhost:6333" # Use local Qdrant server
}
test_vars.update({k: str(v) for k, v in test_dirs.items()})
os.environ.update(test_vars)
yield test_vars
# Clean up test directories
for dir_path in test_dirs.values():
if dir_path.exists():
shutil.rmtree(dir_path, ignore_errors=True)
# Restore original environment
os.environ.clear()
os.environ.update(original_env)
@pytest.fixture
def test_collection_name() -> str:
"""Generate a unique test collection name."""
return f"test_collection_{uuid.uuid4().hex[:8]}"
@pytest_asyncio.fixture
async def qdrant_client() -> QdrantClient:
"""Create a Qdrant client for tests."""
client = QdrantClient(url="http://localhost:6333")
yield client
client.close()
@pytest.mark.asyncio
async def test_server_config_from_env(env_vars, tmp_path, test_collection_name: str, qdrant_client: QdrantClient):
"""Test server configuration from environment variables."""
config = ServerConfig(
host=env_vars["MCP_HOST"],
port=int(env_vars["MCP_PORT"]),
log_level=env_vars["MCP_LOG_LEVEL"],
debug_mode=env_vars["MCP_DEBUG"].lower() == "true",
docs_cache_dir=Path(env_vars["MCP_DOCS_CACHE_DIR"]),
adr_dir=Path(env_vars["MCP_ADR_DIR"]),
kb_storage_dir=Path(env_vars["MCP_KB_STORAGE_DIR"]),
disk_cache_dir=Path(env_vars["MCP_DISK_CACHE_DIR"]),
qdrant_url=env_vars["MCP_QDRANT_URL"],
collection_name=test_collection_name
)
# Create test collection
try:
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
qdrant_client.create_collection(
collection_name=test_collection_name,
vectors_config=VectorParams(
size=384, # Default size for all-MiniLM-L6-v2
distance=Distance.COSINE
)
)
server = CodebaseAnalysisServer(config)
await server.initialize()
assert server.config.host == env_vars["MCP_HOST"]
assert server.config.port == int(env_vars["MCP_PORT"])
assert server.config.log_level == env_vars["MCP_LOG_LEVEL"]
assert server.config.debug_mode == (env_vars["MCP_DEBUG"].lower() == "true")
assert isinstance(server.config.docs_cache_dir, Path)
assert isinstance(server.config.adr_dir, Path)
assert isinstance(server.config.kb_storage_dir, Path)
assert isinstance(server.config.disk_cache_dir, Path)
finally:
await server.shutdown()
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
@pytest.mark.asyncio
async def test_directory_creation(tmp_path, test_collection_name: str, qdrant_client: QdrantClient):
"""Test directory creation."""
config = ServerConfig(
host="localhost",
port=8000,
docs_cache_dir=tmp_path / "docs",
adr_dir=tmp_path / "docs/adrs",
kb_storage_dir=tmp_path / "knowledge",
disk_cache_dir=tmp_path / "cache",
qdrant_url="http://localhost:6333",
collection_name=test_collection_name,
cache_enabled=True # Explicitly enable cache for clarity
)
# Create test collection
try:
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
qdrant_client.create_collection(
collection_name=test_collection_name,
vectors_config=VectorParams(
size=384, # Default size for all-MiniLM-L6-v2
distance=Distance.COSINE
)
)
# Create and initialize server
server = CodebaseAnalysisServer(config)
await server.initialize()
# Verify directories were created
assert (tmp_path / "docs").exists(), "Docs directory was not created"
assert (tmp_path / "docs/adrs").exists(), "ADR directory was not created"
assert (tmp_path / "knowledge").exists(), "Knowledge directory was not created"
assert (tmp_path / "cache").exists(), "Cache directory was not created"
finally:
await server.shutdown()
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
@pytest.mark.asyncio
async def test_directory_creation_with_none_cache_dir(tmp_path, test_collection_name: str, qdrant_client: QdrantClient):
"""Test server startup with None disk_cache_dir."""
config = ServerConfig(
host="localhost",
port=8000,
docs_cache_dir=tmp_path / "docs",
adr_dir=tmp_path / "docs/adrs",
kb_storage_dir=tmp_path / "knowledge",
disk_cache_dir=None, # Explicitly set to None
qdrant_url="http://localhost:6333",
collection_name=test_collection_name,
cache_enabled=True # But keep cache enabled
)
# Create test collection
try:
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
qdrant_client.create_collection(
collection_name=test_collection_name,
vectors_config=VectorParams(
size=384, # Default size for all-MiniLM-L6-v2
distance=Distance.COSINE
)
)
# Initialize server
server = CodebaseAnalysisServer(config)
await server.initialize()
# When disk_cache_dir is None but cache is enabled, we should default to Path("cache")
assert config.disk_cache_dir == Path("cache"), "disk_cache_dir should default to 'cache'"
assert Path("cache").exists(), "Default cache directory should exist"
finally:
await server.shutdown()
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
@pytest.mark.asyncio
async def test_directory_creation_with_cache_disabled(tmp_path, test_collection_name: str, qdrant_client: QdrantClient):
"""Test server startup with caching disabled."""
config = ServerConfig(
host="localhost",
port=8000,
docs_cache_dir=tmp_path / "docs",
adr_dir=tmp_path / "docs/adrs",
kb_storage_dir=tmp_path / "knowledge",
disk_cache_dir=Path(tmp_path / "cache"), # Set a path
qdrant_url="http://localhost:6333",
collection_name=test_collection_name,
cache_enabled=False # But disable caching
)
# Create test collection
try:
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
qdrant_client.create_collection(
collection_name=test_collection_name,
vectors_config=VectorParams(
size=384, # Default size for all-MiniLM-L6-v2
distance=Distance.COSINE
)
)
# Server initialization should set disk_cache_dir to None when cache_enabled is False
server = CodebaseAnalysisServer(config)
await server.initialize()
# Verify that disk_cache_dir is None when cache_enabled is False
assert config.disk_cache_dir is None, "disk_cache_dir should be None when cache_enabled is False"
# And that the cache directory does not exist
assert not (tmp_path / "cache").exists(), "Cache directory should not exist when cache is disabled"
finally:
await server.shutdown()
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
@pytest.mark.asyncio
async def test_directory_creation_permission_error(tmp_path, test_collection_name: str, qdrant_client: QdrantClient):
"""Test directory creation with permission error."""
readonly_dir = tmp_path / "readonly"
readonly_dir.mkdir()
readonly_dir.chmod(0o444) # Read-only
config = ServerConfig(
host="localhost",
port=8000,
docs_cache_dir=readonly_dir / "docs",
adr_dir=readonly_dir / "docs/adrs",
kb_storage_dir=readonly_dir / "knowledge",
disk_cache_dir=readonly_dir / "cache",
qdrant_url="http://localhost:6333",
collection_name=test_collection_name
)
server = None
try:
# Create test collection
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
qdrant_client.create_collection(
collection_name=test_collection_name,
vectors_config=VectorParams(
size=384, # Default size for all-MiniLM-L6-v2
distance=Distance.COSINE
)
)
server = CodebaseAnalysisServer(config)
with pytest.raises(RuntimeError) as exc_info:
await server.initialize()
assert "Permission denied" in str(exc_info.value)
finally:
if server:
await server.shutdown()
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
# Clean up the readonly directory
readonly_dir.chmod(0o777) # Restore write permissions for cleanup
if readonly_dir.exists():
shutil.rmtree(readonly_dir)
@pytest.mark.asyncio
async def test_directory_already_exists(tmp_path, test_collection_name: str, qdrant_client: QdrantClient):
"""Test server initialization with pre-existing directories."""
# Create directories before server initialization
dirs = [
tmp_path / "docs",
tmp_path / "docs/adrs",
tmp_path / "knowledge",
tmp_path / "cache"
]
for dir_path in dirs:
dir_path.mkdir(parents=True, exist_ok=True)
config = ServerConfig(
host="localhost",
port=8000,
docs_cache_dir=tmp_path / "docs",
adr_dir=tmp_path / "docs/adrs",
kb_storage_dir=tmp_path / "knowledge",
disk_cache_dir=tmp_path / "cache",
qdrant_url="http://localhost:6333",
collection_name=test_collection_name
)
# Create test collection
try:
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
qdrant_client.create_collection(
collection_name=test_collection_name,
vectors_config=VectorParams(
size=384, # Default size for all-MiniLM-L6-v2
distance=Distance.COSINE
)
)
server = CodebaseAnalysisServer(config)
await server.initialize()
# Verify directories still exist and are accessible
for dir_path in dirs:
assert dir_path.exists()
assert os.access(dir_path, os.R_OK | os.W_OK)
finally:
await server.shutdown()
if test_collection_name in [c.name for c in qdrant_client.get_collections().collections]:
qdrant_client.delete_collection(test_collection_name)
# Clean up
for dir_path in dirs:
if dir_path.exists():
shutil.rmtree(dir_path)
```
--------------------------------------------------------------------------------
/scripts/store_code_relationships.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
Store Code Component Relationships in Vector Database
This script analyzes the codebase to extract relationships between components
and stores them in the vector database for use in build verification.
"""
import os
import sys
import json
import logging
import asyncio
import argparse
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Any, Set, Tuple
import uuid
# Add the project root to the Python path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from src.mcp_codebase_insight.core.vector_store import VectorStore
from src.mcp_codebase_insight.core.embeddings import SentenceTransformerEmbedding
from qdrant_client import QdrantClient
from qdrant_client.http import models as rest
from qdrant_client.http.models import Filter, FieldCondition, MatchValue
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(),
logging.FileHandler(Path('logs/code_relationships.log'))
]
)
logger = logging.getLogger('code_relationships')
class CodeRelationshipAnalyzer:
"""Code relationship analyzer for storing component relationships in vector database."""
def __init__(self, config_path: str = None):
"""Initialize the code relationship analyzer.
Args:
config_path: Path to configuration file (optional)
"""
self.config = self._load_config(config_path)
self.vector_store = None
self.embedder = None
self.dependency_map = {}
self.critical_components = set()
self.source_files = []
def _load_config(self, config_path: str) -> Dict[str, Any]:
"""Load configuration from file or environment variables.
Args:
config_path: Path to configuration file
Returns:
Configuration dictionary
"""
config = {
'qdrant_url': os.environ.get('QDRANT_URL', 'http://localhost:6333'),
'qdrant_api_key': os.environ.get('QDRANT_API_KEY', ''),
'collection_name': os.environ.get('COLLECTION_NAME', 'mcp-codebase-insight'),
'embedding_model': os.environ.get('EMBEDDING_MODEL', 'sentence-transformers/all-MiniLM-L6-v2'),
'source_dirs': ['src'],
'exclude_dirs': ['__pycache__', '.git', '.venv', 'test_env', 'dist', 'build'],
'critical_modules': [
'mcp_codebase_insight.core.vector_store',
'mcp_codebase_insight.core.knowledge',
'mcp_codebase_insight.server'
]
}
# Override with config file if provided
if config_path:
try:
with open(config_path, 'r') as f:
file_config = json.load(f)
config.update(file_config)
except Exception as e:
logger.error(f"Failed to load config from {config_path}: {e}")
return config
async def initialize(self):
"""Initialize the analyzer."""
logger.info("Initializing code relationship analyzer...")
# Initialize embedder
logger.info("Initializing embedder...")
self.embedder = SentenceTransformerEmbedding(model_name=self.config['embedding_model'])
await self.embedder.initialize()
# Initialize vector store
logger.info(f"Connecting to vector store at {self.config['qdrant_url']}...")
self.vector_store = VectorStore(
url=self.config['qdrant_url'],
embedder=self.embedder,
collection_name=self.config['collection_name'],
api_key=self.config.get('qdrant_api_key'),
vector_name="default" # Specify a vector name for the collection
)
await self.vector_store.initialize()
# Set critical components
self.critical_components = set(self.config.get('critical_modules', []))
logger.info("Code relationship analyzer initialized successfully")
def find_source_files(self) -> List[Path]:
"""Find all source files to analyze.
Returns:
List of source file paths
"""
logger.info("Finding source files...")
source_files = []
source_dirs = [Path(dir_name) for dir_name in self.config['source_dirs']]
exclude_dirs = self.config['exclude_dirs']
for source_dir in source_dirs:
if not source_dir.exists():
logger.warning(f"Source directory {source_dir} does not exist")
continue
for root, dirs, files in os.walk(source_dir):
# Skip excluded directories
dirs[:] = [d for d in dirs if d not in exclude_dirs]
for file in files:
if file.endswith('.py'):
source_files.append(Path(root) / file)
logger.info(f"Found {len(source_files)} source files")
self.source_files = source_files
return source_files
def analyze_file_dependencies(self, file_path: Path) -> Dict[str, List[str]]:
"""Analyze dependencies for a single file.
Args:
file_path: Path to the file to analyze
Returns:
Dictionary mapping module name to list of dependencies
"""
dependencies = []
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# Extract imports
lines = content.split('\n')
for line in lines:
line = line.strip()
# Skip comments
if line.startswith('#'):
continue
# Handle import statements
if line.startswith('import ') or ' import ' in line:
if line.startswith('import '):
# Handle "import module" or "import module as alias"
import_part = line[7:].strip()
if ' as ' in import_part:
import_part = import_part.split(' as ')[0].strip()
dependencies.append(import_part)
elif line.startswith('from ') and ' import ' in line:
# Handle "from module import something"
from_part = line[5:].split(' import ')[0].strip()
dependencies.append(from_part)
# Convert file path to module name
module_name = str(file_path).replace('/', '.').replace('\\', '.').replace('.py', '')
for source_dir in self.config['source_dirs']:
prefix = f"{source_dir}."
if module_name.startswith(prefix):
module_name = module_name[len(prefix):]
return {module_name: dependencies}
except Exception as e:
logger.error(f"Error analyzing file {file_path}: {e}")
return {}
def analyze_all_dependencies(self) -> Dict[str, List[str]]:
"""Analyze dependencies for all source files.
Returns:
Dictionary mapping module names to lists of dependencies
"""
logger.info("Analyzing dependencies for all source files...")
if not self.source_files:
self.find_source_files()
dependency_map = {}
for file_path in self.source_files:
file_dependencies = self.analyze_file_dependencies(file_path)
dependency_map.update(file_dependencies)
logger.info(f"Analyzed dependencies for {len(dependency_map)} modules")
self.dependency_map = dependency_map
return dependency_map
def identify_critical_components(self) -> Set[str]:
"""Identify critical components in the codebase.
Returns:
Set of critical component names
"""
logger.info("Identifying critical components...")
# Start with configured critical modules
critical_components = set(self.critical_components)
# Add modules with many dependents
if self.dependency_map:
# Count how many times each module is a dependency
dependent_count = {}
for module, dependencies in self.dependency_map.items():
for dependency in dependencies:
if dependency in dependent_count:
dependent_count[dependency] += 1
else:
dependent_count[dependency] = 1
# Add modules with more than 3 dependents to critical components
for module, count in dependent_count.items():
if count > 3:
critical_components.add(module)
logger.info(f"Identified {len(critical_components)} critical components")
self.critical_components = critical_components
return critical_components
async def store_in_vector_database(self):
"""Store code relationships in vector database."""
try:
# Store dependency map
dependency_text = json.dumps({
'type': 'dependency_map',
'dependencies': self.dependency_map
})
dependency_vector = await self.vector_store.embedder.embed(dependency_text)
dependency_data = {
'id': str(uuid.uuid4()),
'vector': dependency_vector,
'payload': {
'type': 'dependency_map',
'timestamp': datetime.now().isoformat(),
'module_count': len(self.dependency_map)
}
}
# Store critical components
critical_text = json.dumps({
'type': 'critical_components',
'components': list(self.critical_components)
})
critical_vector = await self.vector_store.embedder.embed(critical_text)
critical_data = {
'id': str(uuid.uuid4()),
'vector': critical_vector,
'payload': {
'type': 'critical_components',
'timestamp': datetime.now().isoformat(),
'component_count': len(self.critical_components)
}
}
# Store build verification criteria
criteria_text = json.dumps({
'type': 'build_criteria',
'critical_modules': list(self.critical_components),
'min_test_coverage': 80.0,
'max_allowed_failures': 0
})
criteria_vector = await self.vector_store.embedder.embed(criteria_text)
criteria_data = {
'id': str(uuid.uuid4()),
'vector': criteria_vector,
'payload': {
'type': 'build_criteria',
'timestamp': datetime.now().isoformat()
}
}
# Store all data points
data_points = [dependency_data, critical_data, criteria_data]
self.vector_store.client.upsert(
collection_name=self.vector_store.collection_name,
points=[rest.PointStruct(
id=data['id'],
vectors={self.vector_store.vector_name: data['vector']},
payload=data['payload']
) for data in data_points]
)
logger.info("Successfully stored code relationships in vector database")
except Exception as e:
logger.error(f"Error storing in vector database: {e}")
raise
async def analyze_and_store(self):
"""Analyze code relationships and store them in the vector database."""
try:
# Find source files
self.find_source_files()
# Analyze dependencies
self.analyze_all_dependencies()
# Identify critical components
self.identify_critical_components()
# Store in vector database
await self.store_in_vector_database()
logger.info("Analysis and storage completed successfully")
return True
except Exception as e:
logger.error(f"Error analyzing and storing code relationships: {e}")
return False
async def cleanup(self):
"""Clean up resources."""
if self.vector_store:
await self.vector_store.cleanup()
await self.vector_store.close()
async def main():
"""Main function."""
parser = argparse.ArgumentParser(description="Code Relationship Analyzer")
parser.add_argument("--config", help="Path to configuration file")
args = parser.parse_args()
# Create logs directory if it doesn't exist
os.makedirs("logs", exist_ok=True)
analyzer = CodeRelationshipAnalyzer(args.config)
try:
await analyzer.initialize()
success = await analyzer.analyze_and_store()
if success:
logger.info("Code relationship analysis completed successfully")
return 0
else:
logger.error("Code relationship analysis failed")
return 1
except Exception as e:
logger.error(f"Error in code relationship analysis: {e}")
return 1
finally:
await analyzer.cleanup()
if __name__ == "__main__":
sys.exit(asyncio.run(main()))
```
--------------------------------------------------------------------------------
/src/mcp_codebase_insight/core/state.py:
--------------------------------------------------------------------------------
```python
"""Server state management."""
from dataclasses import dataclass, field
from typing import Dict, Optional, List, Any, Set
import asyncio
from contextlib import AsyncExitStack
import sys
import threading
from datetime import datetime
import logging
import uuid
from ..utils.logger import get_logger
from .config import ServerConfig
from .di import DIContainer
from .task_tracker import TaskTracker
from .component_status import ComponentStatus
logger = get_logger(__name__)
@dataclass
class ComponentState:
"""State tracking for a server component."""
status: ComponentStatus = ComponentStatus.UNINITIALIZED
error: Optional[str] = None
instance: Any = None
last_update: datetime = field(default_factory=datetime.utcnow)
retry_count: int = 0
instance_id: str = field(default_factory=lambda: str(uuid.uuid4()))
class ServerState:
"""Global server state management."""
def __init__(self):
"""Initialize server state."""
self._init_lock = asyncio.Lock()
self._cleanup_lock = asyncio.Lock()
self.initialized = False
self.config: Optional[ServerConfig] = None
self._components: Dict[str, ComponentState] = {}
self._cleanup_handlers: List[asyncio.Task] = []
self._task_tracker = TaskTracker()
self._instance_id = str(uuid.uuid4())
logger.info(f"Created ServerState instance {self._instance_id}")
def register_component(self, name: str, instance: Any = None) -> None:
"""Register a new component."""
if name not in self._components:
component_state = ComponentState()
if instance:
component_state.instance = instance
self._components[name] = component_state
logger.debug(f"Registered component: {name}")
def update_component_status(
self,
name: str,
status: ComponentStatus,
error: Optional[str] = None,
instance: Any = None
) -> None:
"""Update component status."""
if name not in self._components:
self.register_component(name)
component = self._components[name]
component.status = status
component.error = error
component.last_update = datetime.utcnow()
if instance is not None:
component.instance = instance
if status == ComponentStatus.FAILED:
component.retry_count += 1
logger.debug(
f"Component {name} status updated to {status}"
f"{f' (error: {error})' if error else ''}"
)
def get_component(self, name: str) -> Any:
"""Get component instance."""
if name not in self._components:
logger.warning(f"Component {name} not registered")
return None
component = self._components[name]
if component.status != ComponentStatus.INITIALIZED:
logger.warning(f"Component {name} not initialized (status: {component.status.value})")
return None
return component.instance
def register_background_task(self, task: asyncio.Task) -> None:
"""Register a background task for tracking and cleanup."""
self._task_tracker.track_task(task)
logger.debug(f"Registered background task: {task.get_name()}")
async def cancel_background_tasks(self) -> None:
"""Cancel all tracked background tasks."""
await self._task_tracker.cancel_all_tasks()
async def cleanup(self) -> None:
"""Cleanup server components."""
async with self._cleanup_lock:
if not self.initialized:
logger.warning("Server not initialized, nothing to clean up")
return
logger.info(f"Beginning cleanup for instance {self._instance_id}")
# First, cancel any background tasks
await self.cancel_background_tasks()
# Clean up components in reverse order
components = list(self._components.keys())
components.reverse()
for component in components:
self.update_component_status(component, ComponentStatus.CLEANING)
try:
# Component-specific cleanup logic here
comp_instance = self._components[component].instance
if comp_instance and hasattr(comp_instance, 'cleanup'):
await comp_instance.cleanup()
self.update_component_status(component, ComponentStatus.CLEANED)
except Exception as e:
error_msg = f"Error cleaning up {component}: {str(e)}"
logger.error(error_msg, exc_info=True)
self.update_component_status(
component,
ComponentStatus.FAILED,
error_msg
)
# Cancel any remaining cleanup handlers
for task in self._cleanup_handlers:
if not task.done():
task.cancel()
self.initialized = False
logger.info(f"Server instance {self._instance_id} cleanup completed")
def get_component_status(self) -> Dict[str, Any]:
"""Get status of all components."""
return {
name: {
"status": comp.status.value,
"error": comp.error,
"last_update": comp.last_update.isoformat(),
"retry_count": comp.retry_count,
"instance_id": comp.instance_id
}
for name, comp in self._components.items()
}
def register_cleanup_handler(self, task: asyncio.Task) -> None:
"""Register a cleanup handler task."""
self._cleanup_handlers.append(task)
logger.debug(f"Registered cleanup handler: {task.get_name()}")
@property
def instance_id(self) -> str:
"""Get the unique instance ID of this server state."""
return self._instance_id
def list_components(self) -> List[str]:
"""List all registered components."""
return list(self._components.keys())
def get_active_tasks(self) -> Set[asyncio.Task]:
"""Get all currently active tasks."""
return self._task_tracker.get_active_tasks()
def get_task_count(self) -> int:
"""Get the number of currently tracked tasks."""
return self._task_tracker.get_task_count()
async def initialize(self) -> None:
"""Initialize server components."""
async with self._init_lock:
if self.initialized:
logger.warning("Server already initialized")
return
logger.info(f"Beginning initialization for instance {self._instance_id}")
try:
# Initialize components in order
components = [
"database",
"vector_store",
"task_manager",
"analysis_engine",
"adr_manager",
"knowledge_base",
"mcp_server"
]
for component in components:
self.update_component_status(component, ComponentStatus.INITIALIZING)
try:
# Component-specific initialization logic here
# await self._initialize_component(component)
# For now, let's just mark them as initialized
# In a real implementation, you'd create and store the actual component instances
# For the vector_store component, create a real instance
if component == "vector_store":
from .vector_store import VectorStore
from .embeddings import SentenceTransformerEmbedding
# If config is available, use it to configure the vector store
if self.config:
embedder = SentenceTransformerEmbedding(self.config.embedding_model)
vector_store = VectorStore(
url=self.config.qdrant_url,
embedder=embedder,
collection_name=self.config.collection_name
)
await vector_store.initialize()
self.update_component_status(
"vector_store",
ComponentStatus.INITIALIZED,
instance=vector_store
)
# For the adr_manager component
elif component == "adr_manager":
from .adr import ADRManager
if self.config:
adr_manager = ADRManager(self.config)
await adr_manager.initialize()
self.update_component_status(
"adr_manager",
ComponentStatus.INITIALIZED,
instance=adr_manager
)
# For the knowledge_base component
elif component == "knowledge_base":
from .knowledge import KnowledgeBase
if self.config:
# Get vector_store if available
vector_store = self.get_component("vector_store")
if vector_store:
kb = KnowledgeBase(self.config, vector_store)
await kb.initialize()
self.update_component_status(
"knowledge_base",
ComponentStatus.INITIALIZED,
instance=kb
)
else:
error_msg = "Vector store not initialized, cannot initialize knowledge base"
logger.error(error_msg)
self.update_component_status(
component,
ComponentStatus.FAILED,
error=error_msg
)
# For task_manager component
elif component == "task_manager":
from .tasks import TaskManager
if self.config:
task_manager = TaskManager(self.config)
await task_manager.initialize()
self.update_component_status(
"task_manager",
ComponentStatus.INITIALIZED,
instance=task_manager
)
# For database component (placeholder)
elif component == "database":
# Mock implementation for database
self.update_component_status(
"database",
ComponentStatus.INITIALIZED,
instance={"status": "mocked"}
)
# For analysis_engine component (placeholder)
elif component == "analysis_engine":
# Mock implementation for analysis engine
self.update_component_status(
"analysis_engine",
ComponentStatus.INITIALIZED,
instance={"status": "mocked"}
)
# For mcp_server component (placeholder)
elif component == "mcp_server":
# Mock implementation for mcp server
self.update_component_status(
"mcp_server",
ComponentStatus.INITIALIZED,
instance={"status": "mocked"}
)
except Exception as e:
error_msg = f"Failed to initialize {component}: {str(e)}"
logger.error(error_msg, exc_info=True)
self.update_component_status(
component,
ComponentStatus.FAILED,
error=error_msg
)
# Set server as initialized if all critical components are initialized
critical_components = ["vector_store", "task_manager", "mcp_server"]
all_critical_initialized = all(
self._components.get(c) and
self._components[c].status == ComponentStatus.INITIALIZED
for c in critical_components
)
if all_critical_initialized:
self.initialized = True
logger.info(f"Server instance {self._instance_id} initialized successfully")
else:
logger.warning(
f"Server instance {self._instance_id} partially initialized "
f"(some critical components failed)"
)
except Exception as e:
error_msg = f"Failed to initialize server: {str(e)}"
logger.error(error_msg, exc_info=True)
raise
```
--------------------------------------------------------------------------------
/.github/workflows/build-verification.yml:
--------------------------------------------------------------------------------
```yaml
name: Build Verification
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
workflow_dispatch:
inputs:
config_file:
description: 'Path to verification config file'
required: false
default: 'verification-config.json'
min_coverage:
description: 'Minimum test coverage percentage'
required: false
default: '80.0'
max_failures:
description: 'Maximum allowed test failures'
required: false
default: '0'
python_version:
description: 'Python version to use for verification'
required: false
default: '3.9'
jobs:
verify:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: [ '3.10', '3.11', '3.12', '3.13' ]
fail-fast: false # Continue testing other Python versions even if one fails
name: Verify with Python ${{ matrix.python-version }}
environment:
name: production
url: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
services:
qdrant:
image: qdrant/qdrant:v1.13.6
ports:
- 6333:6333
- 6334:6334
steps:
- name: Checkout code
uses: actions/checkout@v4
with:
fetch-depth: 0 # Fetch all history for dependencies analysis
- name: Set up Python ${{ matrix.python-version }}
uses: actions/[email protected]
with:
python-version: ${{ matrix.python-version }}
cache: 'pip'
- name: Wait for Qdrant and verify connection
run: |
echo "Waiting for Qdrant to start..."
chmod +x scripts/check_qdrant_health.sh
./scripts/check_qdrant_health.sh "http://localhost:6333" 20 5
- name: Setup private packages
run: |
# Create local-packages directory if it doesn't exist
mkdir -p local-packages
# If there are private packages in repositories, clone them here
if [ -n "${{ secrets.PRIVATE_REPO_URL }}" ]; then
echo "Setting up private package repository..."
# Configure pip to use the private repository if provided
mkdir -p ~/.pip
echo "[global]" > ~/.pip/pip.conf
echo "index-url = https://pypi.org/simple" >> ~/.pip/pip.conf
# Add the private repository with token if available
if [ -n "${{ secrets.PRIVATE_REPO_TOKEN }}" ]; then
echo "extra-index-url = ${{ secrets.PRIVATE_REPO_URL }}:${{ secrets.PRIVATE_REPO_TOKEN }}@simple" >> ~/.pip/pip.conf
else
echo "extra-index-url = ${{ secrets.PRIVATE_REPO_URL }}/simple" >> ~/.pip/pip.conf
fi
fi
# If there are local Git repositories for dependencies, clone them
if [ -n "${{ secrets.MCP_SERVER_QDRANT_REPO }}" ]; then
echo "Cloning mcp-server-qdrant from repository..."
git clone "${{ secrets.MCP_SERVER_QDRANT_REPO }}" local-packages/mcp-server-qdrant
# Install the package in development mode
cd local-packages/mcp-server-qdrant
pip install -e .
cd ../../
fi
# Similarly for uvx package if needed
if [ -n "${{ secrets.UVX_REPO }}" ]; then
echo "Cloning uvx from repository..."
git clone "${{ secrets.UVX_REPO }}" local-packages/uvx
# Install the package in development mode
cd local-packages/uvx
pip install -e .
cd ../../
fi
- name: Install dependencies
run: |
python -m pip install --upgrade pip setuptools wheel
# Make the requirements script executable
chmod +x scripts/compile_requirements.sh
# Set environment variables for private package handling
export PRIVATE_REPO_URL="${{ secrets.PRIVATE_REPO_URL }}"
export PRIVATE_REPO_TOKEN="${{ secrets.PRIVATE_REPO_TOKEN }}"
export LOCAL_PACKAGE_PATHS="./local-packages"
# Use the compile_requirements.sh script to generate version-specific requirements
echo "Using compile_requirements.sh to generate dependencies for Python ${{ matrix.python-version }}..."
# Set auto-yes for cleanup to avoid interactive prompts in CI
echo "y" | ./scripts/compile_requirements.sh ${{ matrix.python-version }}
# Install the generated requirements
if [ -f requirements-${{ matrix.python-version }}.txt ]; then
echo "Installing from version-specific requirements file..."
pip install -r requirements-${{ matrix.python-version }}.txt
pip install -r requirements-dev.txt
# Install private packages if they're in a separate file
if [ -f requirements-private-${{ matrix.python-version }}.txt ]; then
echo "Installing private packages..."
# Try to install private packages, but continue even if it fails
pip install -r requirements-private-${{ matrix.python-version }}.txt || echo "Warning: Some private packages could not be installed"
fi
else
echo "Version-specific requirements not found, falling back to standard requirements.txt"
pip install -r requirements.txt || {
echo "Error installing from requirements.txt, attempting to fix compatibility issues..."
grep -v "^#" requirements.txt | cut -d= -f1 | xargs pip install
}
fi
# Install the package in development mode
pip install -e .
- name: Set up environment
run: |
# Create required directories
mkdir -p logs knowledge cache
{
echo "QDRANT_URL=http://localhost:6333"
echo "MCP_QDRANT_URL=http://localhost:6333"
echo "COLLECTION_NAME=mcp-codebase-insight-${{ matrix.python-version }}"
echo "MCP_COLLECTION_NAME=mcp-codebase-insight-${{ matrix.python-version }}"
echo "EMBEDDING_MODEL=sentence-transformers/all-MiniLM-L6-v2"
echo "BUILD_COMMAND=make build"
echo "TEST_COMMAND=make test"
echo "MIN_TEST_COVERAGE=${{ github.event.inputs.min_coverage || '40.0' }}"
echo "MAX_ALLOWED_FAILURES=${{ github.event.inputs.max_failures || '0' }}"
echo "CRITICAL_MODULES=mcp_codebase_insight.core.vector_store,mcp_codebase_insight.core.knowledge,mcp_codebase_insight.server"
echo "PYTHON_VERSION=${{ matrix.python-version }}"
} >> "$GITHUB_ENV"
- name: Initialize Qdrant collection
run: |
echo "Creating Qdrant collection for testing..."
# Create a basic Python script to initialize the collection
cat > init_qdrant.py << 'EOF'
import os
from qdrant_client import QdrantClient
from qdrant_client.http import models
# Connect to Qdrant
client = QdrantClient(url="http://localhost:6333")
collection_name = os.environ.get("COLLECTION_NAME", "mcp-codebase-insight-${{ matrix.python-version }}")
# Check if collection exists
collections = client.get_collections().collections
collection_names = [c.name for c in collections]
if collection_name in collection_names:
print(f"Collection {collection_name} already exists, recreating it...")
client.delete_collection(collection_name=collection_name)
# Create collection with vector size 384 (for all-MiniLM-L6-v2)
client.create_collection(
collection_name=collection_name,
vectors_config=models.VectorParams(
size=384, # Dimension for all-MiniLM-L6-v2
distance=models.Distance.COSINE,
),
)
print(f"Successfully created collection {collection_name}")
EOF
# Run the initialization script
python init_qdrant.py
# Verify the collection was created
curl -s "http://localhost:6333/collections/$COLLECTION_NAME" || (echo "Failed to create Qdrant collection" && exit 1)
echo "Qdrant collection initialized successfully."
- name: Create configuration file
if: ${{ github.event.inputs.config_file != '' }}
run: |
cat > ${{ github.event.inputs.config_file }} << EOF
{
"success_criteria": {
"min_test_coverage": ${{ github.event.inputs.min_coverage || '40.0' }},
"max_allowed_failures": ${{ github.event.inputs.max_failures || '0' }},
"critical_modules": ["mcp_codebase_insight.core.vector_store", "mcp_codebase_insight.core.knowledge", "mcp_codebase_insight.server"],
"performance_threshold_ms": 500
}
}
EOF
- name: Run build verification
id: verify-build
run: |
# Run specific tests that are known to pass
echo "Running specific tests that are known to pass..."
python -m pytest \
tests/components/test_core_components.py::test_adr_manager \
tests/components/test_sse_components.py::test_get_starlette_app \
tests/components/test_sse_components.py::test_create_sse_server \
tests/components/test_sse_components.py::test_vector_search_tool \
tests/components/test_sse_components.py::test_knowledge_search_tool \
tests/components/test_sse_components.py::test_adr_list_tool \
tests/components/test_sse_components.py::test_task_status_tool \
tests/components/test_sse_components.py::test_sse_handle_connect \
tests/components/test_stdio_components.py::test_stdio_registration \
tests/components/test_stdio_components.py::test_stdio_message_streaming \
tests/components/test_stdio_components.py::test_stdio_error_handling \
tests/components/test_stdio_components.py::test_stdio_large_message \
tests/components/test_knowledge_base.py \
tests/integration/test_server.py::test_vector_store_search_threshold_validation \
tests/integration/test_server.py::test_vector_store_search_functionality \
tests/integration/test_server.py::test_vector_store_search_error_handling \
tests/integration/test_server.py::test_vector_store_search_performance \
tests/integration/test_api_endpoints.py::test_health_check \
tests/integration/test_api_endpoints.py::test_endpoint_integration \
tests/integration/test_api_endpoints.py::test_error_handling \
tests/integration/test_communication_integration.py::test_sse_stdio_interaction \
tests/test_file_relationships.py \
-v -p pytest_asyncio --cov=src/mcp_codebase_insight --cov-report=xml:coverage.xml --cov-report=html:htmlcov
TEST_EXIT_CODE=$?
CONFIG_ARG=""
# Use config file if it exists and is not empty
if [ -n "${{ github.event.inputs.config_file }}" ] && [ -f "${{ github.event.inputs.config_file }}" ] && [ -s "${{ github.event.inputs.config_file }}" ]; then
CONFIG_ARG="--config ${{ github.event.inputs.config_file }}"
python -m scripts.verify_build $CONFIG_ARG --output build-verification-report.json
else
python -m scripts.verify_build --output build-verification-report.json
fi
VERIFY_EXIT_CODE=$?
# Use new output syntax
if [ $TEST_EXIT_CODE -ne 0 ] || [ $VERIFY_EXIT_CODE -ne 0 ]; then
echo "failed=true" >> "$GITHUB_OUTPUT"
fi
- name: Upload verification report
uses: actions/upload-artifact@v4
with:
name: build-verification-report
path: build-verification-report.json
- name: Parse verification report
id: parse-report
if: always()
run: |
if [ -f build-verification-report.json ]; then
SUMMARY=$(jq -r '.build_verification_report.summary' build-verification-report.json)
echo "summary=$SUMMARY" >> "$GITHUB_OUTPUT"
STATUS=$(jq -r '.build_verification_report.verification_results.overall_status' build-verification-report.json)
echo "status=$STATUS" >> "$GITHUB_OUTPUT"
{
echo "## Build Verification Report"
echo "### Status: $STATUS"
echo "### Summary: $SUMMARY"
echo "### Test Results"
TOTAL=$(jq -r '.build_verification_report.test_summary.total' build-verification-report.json)
PASSED=$(jq -r '.build_verification_report.test_summary.passed' build-verification-report.json)
FAILED=$(jq -r '.build_verification_report.test_summary.failed' build-verification-report.json)
COVERAGE=$(jq -r '.build_verification_report.test_summary.coverage' build-verification-report.json)
echo "- Total Tests: $TOTAL"
echo "- Passed: $PASSED"
echo "- Failed: $FAILED"
echo "- Coverage: $COVERAGE%"
} > report.md
if jq -e '.build_verification_report.failure_analysis' build-verification-report.json > /dev/null; then
{
echo "### Failures Detected"
jq -r '.build_verification_report.failure_analysis[] | "- " + .description' build-verification-report.json
} >> report.md
fi
if jq -e '.build_verification_report.contextual_verification' build-verification-report.json > /dev/null; then
{
echo "### Contextual Analysis"
jq -r '.build_verification_report.contextual_verification[] | "#### Module: " + .module + "\n- Failure: " + .failure + "\n- Dependencies: " + (.dependencies | join(", ")) + "\n\n**Potential Causes:**\n" + (.potential_causes | map("- " + .) | join("\n")) + "\n\n**Recommended Actions:**\n" + (.recommended_actions | map("- " + .) | join("\n"))' build-verification-report.json
} >> report.md
fi
else
{
echo "summary=Build verification failed - no report generated" >> "$GITHUB_OUTPUT"
echo "status=FAILED" >> "$GITHUB_OUTPUT"
echo "## Build Verification Failed"
echo "No report was generated. Check the logs for more information."
} > report.md
fi
cat report.md
- name: Create GitHub check
uses: LouisBrunner/[email protected]
if: always()
with:
token: ${{ secrets.GITHUB_TOKEN }}
name: Build Verification
conclusion: ${{ steps.parse-report.outputs.status == 'PASS' && 'success' || 'failure' }}
output: |
{
"title": "Build Verification Results",
"summary": "${{ steps.parse-report.outputs.summary }}",
"text": "${{ steps.parse-report.outputs.report }}"
}
- name: Check verification status
if: steps.verify-build.outputs.failed == 'true' || steps.parse-report.outputs.status != 'PASS'
run: |
echo "Build verification failed!"
exit 1
```
--------------------------------------------------------------------------------
/src/mcp_codebase_insight/core/tasks.py:
--------------------------------------------------------------------------------
```python
"""Task management module."""
import asyncio
from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional
from uuid import UUID, uuid4
import json
from pathlib import Path
from pydantic import BaseModel
class TaskType(str, Enum):
"""Task type enumeration."""
CODE_ANALYSIS = "code_analysis"
PATTERN_EXTRACTION = "pattern_extraction"
DOCUMENTATION = "documentation"
DOCUMENTATION_CRAWL = "doc_crawl"
DEBUG = "debug"
ADR = "adr"
class TaskStatus(str, Enum):
"""Task status enumeration."""
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class TaskPriority(str, Enum):
"""Task priority enumeration."""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class Task(BaseModel):
"""Task model."""
id: UUID
type: TaskType
title: str
description: str
status: TaskStatus
priority: TaskPriority
context: Dict
result: Optional[Dict] = None
error: Optional[str] = None
created_at: datetime
updated_at: datetime
completed_at: Optional[datetime] = None
metadata: Optional[Dict[str, str]] = None
class TaskManager:
"""Manager for asynchronous tasks."""
def __init__(
self,
config,
adr_manager=None,
debug_system=None,
doc_manager=None,
knowledge_base=None,
prompt_manager=None
):
"""Initialize task manager."""
self.config = config
self.adr_manager = adr_manager
self.debug_system = debug_system
self.doc_manager = doc_manager
self.kb = knowledge_base
self.prompt_manager = prompt_manager
# Initialize tasks directory
self.tasks_dir = Path(config.docs_cache_dir) / "tasks"
self.tasks_dir.mkdir(parents=True, exist_ok=True)
self.tasks: Dict[UUID, Task] = {}
self.task_queue: asyncio.Queue = asyncio.Queue()
self.running = False
self._process_task_future = None
self.initialized = False
async def initialize(self):
"""Initialize task manager and start processing tasks."""
if self.initialized:
return
try:
# Create a fresh queue
self.task_queue = asyncio.Queue()
# Load existing tasks from disk
if self.tasks_dir.exists():
for task_file in self.tasks_dir.glob("*.json"):
try:
with open(task_file) as f:
data = json.load(f)
task = Task(**data)
self.tasks[task.id] = task
except Exception as e:
print(f"Error loading task {task_file}: {e}")
# Start task processing
await self.start()
self.initialized = True
except Exception as e:
print(f"Error initializing task manager: {e}")
await self.cleanup()
raise RuntimeError(f"Failed to initialize task manager: {str(e)}")
async def cleanup(self):
"""Clean up task manager and stop processing tasks."""
if not self.initialized:
return
try:
# Stop task processing
await self.stop()
# Save any remaining tasks
for task in self.tasks.values():
if task.status == TaskStatus.IN_PROGRESS:
task.status = TaskStatus.FAILED
task.error = "Server shutdown"
task.updated_at = datetime.utcnow()
await self._save_task(task)
except Exception as e:
print(f"Error cleaning up task manager: {e}")
finally:
self.initialized = False
async def start(self):
"""Start task processing."""
if not self.running:
self.running = True
self._process_task_future = asyncio.create_task(self._process_tasks())
async def stop(self):
"""Stop task processing."""
if self.running:
self.running = False
if self._process_task_future:
try:
# Wait for the task to finish with a timeout
await asyncio.wait_for(self._process_task_future, timeout=5.0)
except asyncio.TimeoutError:
# If it doesn't finish in time, cancel it
self._process_task_future.cancel()
try:
await self._process_task_future
except asyncio.CancelledError:
pass
finally:
self._process_task_future = None
# Create a new empty queue instead of trying to drain the old one
# This avoids task_done() issues
self.task_queue = asyncio.Queue()
async def _save_task(self, task: Task):
"""Save task to disk."""
task_path = self.tasks_dir / f"{task.id}.json"
with open(task_path, "w") as f:
json.dump(task.model_dump(), f, indent=2, default=str)
async def create_task(
self,
type: str,
title: str,
description: str,
context: Dict,
priority: TaskPriority = TaskPriority.MEDIUM,
metadata: Optional[Dict[str, str]] = None
) -> Task:
"""Create a new task."""
now = datetime.utcnow()
task = Task(
id=uuid4(),
type=TaskType(type),
title=title,
description=description,
status=TaskStatus.PENDING,
priority=priority,
context=context,
metadata=metadata,
created_at=now,
updated_at=now
)
self.tasks[task.id] = task
await self._save_task(task) # Save task to disk
await self.task_queue.put(task)
return task
async def get_task(self, task_id: str) -> Optional[Task]:
"""Get task by ID."""
task_path = self.tasks_dir / f"{task_id}.json"
if not task_path.exists():
return None
with open(task_path) as f:
data = json.load(f)
return Task(**data)
async def update_task(
self,
task_id: str,
status: Optional[str] = None,
result: Optional[Dict] = None,
error: Optional[str] = None
) -> Optional[Task]:
"""Update task status and result."""
task = await self.get_task(task_id)
if not task:
return None
if status:
task.status = status
if result:
task.result = result
if error:
task.error = error
task.updated_at = datetime.utcnow()
if status == "completed":
task.completed_at = datetime.utcnow()
await self._save_task(task)
return task
async def cancel_task(self, task_id: UUID) -> Optional[Task]:
"""Cancel a pending or in-progress task."""
task = self.tasks.get(task_id)
if not task:
return None
if task.status in [TaskStatus.PENDING, TaskStatus.IN_PROGRESS]:
task.status = TaskStatus.CANCELLED
task.updated_at = datetime.utcnow()
return task
async def list_tasks(
self,
type: Optional[TaskType] = None,
status: Optional[TaskStatus] = None,
priority: Optional[TaskPriority] = None
) -> List[Task]:
"""List all tasks, optionally filtered."""
tasks = []
for task in self.tasks.values():
if type and task.type != type:
continue
if status and task.status != status:
continue
if priority and task.priority != priority:
continue
tasks.append(task)
return sorted(tasks, key=lambda x: x.created_at)
async def _process_tasks(self):
"""Process tasks from queue."""
while self.running:
try:
# Use get with timeout to avoid blocking forever
try:
task = await asyncio.wait_for(self.task_queue.get(), timeout=1.0)
except asyncio.TimeoutError:
continue
# Update status
task.status = TaskStatus.IN_PROGRESS
task.updated_at = datetime.utcnow()
try:
# Process task based on type
if task.type == TaskType.CODE_ANALYSIS:
await self._process_code_analysis(task)
elif task.type == TaskType.PATTERN_EXTRACTION:
result = await self._extract_patterns(task)
elif task.type == TaskType.DOCUMENTATION:
result = await self._generate_documentation(task)
elif task.type == TaskType.DOCUMENTATION_CRAWL:
result = await self._crawl_documentation(task)
elif task.type == TaskType.DEBUG:
result = await self._debug_issue(task)
elif task.type == TaskType.ADR:
result = await self._process_adr(task)
else:
raise ValueError(f"Unknown task type: {task.type}")
# Update task with result
task.result = result
task.status = TaskStatus.COMPLETED
except Exception as e:
# Update task with error
task.error = str(e)
task.status = TaskStatus.FAILED
task.completed_at = datetime.utcnow()
task.updated_at = datetime.utcnow()
# Mark task as done in the queue
self.task_queue.task_done()
except asyncio.CancelledError:
# Don't call task_done() here since we didn't get a task
break
except Exception as e:
# Log error but continue processing
print(f"Error processing task: {e}")
# Don't call task_done() here since we might not have gotten a task
async def _process_code_analysis(self, task: Task) -> None:
"""Process a code analysis task."""
try:
code = task.context.get("code", "")
context = task.context.get("context", {})
patterns = await self.app.state.knowledge.analyze_code(
code=code,
language=context.get("language", "python"),
purpose=context.get("purpose", "")
)
await self._update_task(
task,
status=TaskStatus.COMPLETED,
result={"patterns": [p.pattern.model_dump() for p in patterns]}
)
except Exception as e:
self.logger.error(f"Failed to process code analysis task: {str(e)}")
await self._update_task(
task,
status=TaskStatus.FAILED,
error=str(e)
)
async def _extract_patterns(self, task: Task) -> Dict:
"""Extract patterns from code."""
if not self.kb:
raise ValueError("Knowledge base not available")
code = task.context.get("code")
if not code:
raise ValueError("No code provided for pattern extraction")
# TODO: Implement pattern extraction logic
return {
"patterns": []
}
async def _generate_documentation(self, task: Task) -> Dict:
"""Generate documentation."""
if not self.doc_manager:
raise ValueError("Documentation manager not available")
content = task.context.get("content")
if not content:
raise ValueError("No content provided for documentation")
doc = await self.doc_manager.add_document(
title=task.title,
content=content,
type="documentation",
metadata=task.metadata
)
return {
"document_id": str(doc.id),
"path": f"docs/{doc.id}.json"
}
async def _crawl_documentation(self, task: Task) -> Dict:
"""Crawl documentation from URLs."""
if not self.doc_manager:
raise ValueError("Documentation manager not available")
urls = task.context.get("urls")
source_type = task.context.get("source_type")
if not urls or not source_type:
raise ValueError("Missing required fields: urls, source_type")
docs = await self.doc_manager.crawl_docs(
urls=urls,
source_type=source_type
)
return {
"documents": [doc.model_dump() for doc in docs],
"total_documents": len(docs)
}
async def _debug_issue(self, task: Task) -> Dict:
"""Debug an issue."""
if not self.debug_system:
raise ValueError("Debug system not available")
issue = await self.debug_system.create_issue(
title=task.title,
type="bug",
description=task.context
)
steps = await self.debug_system.analyze_issue(issue.id)
return {
"issue_id": str(issue.id),
"steps": steps
}
async def _process_adr(self, task: Task) -> Dict:
"""Process ADR-related task."""
if not self.adr_manager:
raise ValueError("ADR manager not available")
adr = await self.adr_manager.create_adr(
title=task.title,
context=task.context.get("context", {}),
options=task.context.get("options", []),
decision=task.context.get("decision", "")
)
return {
"adr_id": str(adr.id),
"path": f"docs/adrs/{adr.id}.json"
}
async def _process_doc_crawl(self, task: Task) -> None:
"""Process a document crawl task."""
try:
urls = task.context.get("urls", [])
source_type = task.context.get("source_type", "markdown")
total_documents = 0
for url in urls:
try:
await self.doc_manager.crawl_document(url, source_type)
total_documents += 1
except Exception as e:
print(f"Failed to crawl document {url}: {str(e)}")
task.status = TaskStatus.COMPLETED
task.result = {"total_documents": total_documents}
task.updated_at = datetime.utcnow()
task.completed_at = datetime.utcnow()
await self._save_task(task)
except Exception as e:
print(f"Failed to process doc crawl task: {str(e)}")
task.status = TaskStatus.FAILED
task.error = str(e)
task.updated_at = datetime.utcnow()
await self._save_task(task)
```
--------------------------------------------------------------------------------
/component_test_runner.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""
Component Test Runner
A specialized runner for executing component tests with proper async fixture handling.
This bypasses the standard pytest fixture mechanisms to handle async fixtures correctly
in isolated execution environments.
"""
import os
import sys
import uuid
import asyncio
import importlib
from pathlib import Path
import inspect
import logging
import re
from typing import Dict, Any, List, Callable, Tuple, Optional, Set, Awaitable
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger("component-test-runner")
# Import the sys module to modify path
import sys
sys.path.insert(0, '/Users/tosinakinosho/workspaces/mcp-codebase-insight')
# Import required components directly to avoid fixture resolution issues
from src.mcp_codebase_insight.core.config import ServerConfig
from src.mcp_codebase_insight.core.vector_store import VectorStore
from src.mcp_codebase_insight.core.embeddings import SentenceTransformerEmbedding
from src.mcp_codebase_insight.core.knowledge import KnowledgeBase
from src.mcp_codebase_insight.core.tasks import TaskManager
async def create_test_config() -> ServerConfig:
"""Create a server configuration for tests."""
# Generate a unique collection name for this test run
collection_name = f"test_collection_{uuid.uuid4().hex[:8]}"
# Check if MCP_COLLECTION_NAME is set in env, use that instead if available
if "MCP_COLLECTION_NAME" in os.environ:
collection_name = os.environ["MCP_COLLECTION_NAME"]
logger.info(f"Using test collection: {collection_name}")
config = ServerConfig(
host="localhost",
port=8000,
log_level="DEBUG",
qdrant_url="http://localhost:6333",
docs_cache_dir=Path(".test_cache") / "docs",
adr_dir=Path(".test_cache") / "docs/adrs",
kb_storage_dir=Path(".test_cache") / "knowledge",
embedding_model="all-MiniLM-L6-v2",
collection_name=collection_name,
debug_mode=True,
metrics_enabled=False,
cache_enabled=True,
memory_cache_size=1000,
disk_cache_dir=Path(".test_cache") / "cache"
)
return config
async def create_embedder() -> SentenceTransformerEmbedding:
"""Create an embedder for tests."""
logger.info("Initializing the embedder...")
return SentenceTransformerEmbedding()
async def create_vector_store(config: ServerConfig, embedder: SentenceTransformerEmbedding) -> VectorStore:
"""Create a vector store for tests."""
logger.info("Initializing the vector store...")
store = VectorStore(config.qdrant_url, embedder)
try:
await store.initialize()
logger.info("Vector store initialized successfully")
return store
except Exception as e:
logger.error(f"Failed to initialize vector store: {e}")
raise RuntimeError(f"Failed to initialize vector store: {e}")
async def create_knowledge_base(config: ServerConfig, vector_store: VectorStore) -> KnowledgeBase:
"""Create a knowledge base for tests."""
logger.info("Initializing the knowledge base...")
kb = KnowledgeBase(config, vector_store)
try:
await kb.initialize()
logger.info("Knowledge base initialized successfully")
return kb
except Exception as e:
logger.error(f"Failed to initialize knowledge base: {e}")
raise RuntimeError(f"Failed to initialize knowledge base: {e}")
async def create_task_manager(config: ServerConfig) -> TaskManager:
"""Create a task manager for tests."""
logger.info("Initializing the task manager...")
manager = TaskManager(config)
try:
await manager.initialize()
logger.info("Task manager initialized successfully")
return manager
except Exception as e:
logger.error(f"Failed to initialize task manager: {e}")
raise RuntimeError(f"Failed to initialize task manager: {e}")
async def create_test_metadata() -> Dict[str, Any]:
"""Standard test metadata for consistency across tests."""
return {
"type": "code",
"language": "python",
"title": "Test Code",
"description": "Test code snippet for vector store testing",
"tags": ["test", "vector"]
}
def create_test_code() -> str:
"""Provide sample code for testing task-related functionality."""
return """
def example_function():
\"\"\"This is a test function for task manager tests.\"\"\"
return "Hello, world!"
class TestClass:
def __init__(self):
self.value = 42
def method(self):
return self.value
"""
async def cleanup_vector_store(vector_store: VectorStore) -> None:
"""Cleanup a vector store after tests."""
if vector_store and hasattr(vector_store, 'cleanup'):
logger.info("Cleaning up vector store...")
try:
await vector_store.cleanup()
logger.info("Vector store cleanup completed")
except Exception as e:
logger.error(f"Error during vector store cleanup: {e}")
async def cleanup_knowledge_base(kb: KnowledgeBase) -> None:
"""Cleanup a knowledge base after tests."""
if kb and hasattr(kb, 'cleanup'):
logger.info("Cleaning up knowledge base...")
try:
await kb.cleanup()
logger.info("Knowledge base cleanup completed")
except Exception as e:
logger.error(f"Error during knowledge base cleanup: {e}")
async def cleanup_task_manager(manager: TaskManager) -> None:
"""Cleanup a task manager after tests."""
if manager and hasattr(manager, 'cleanup'):
logger.info("Cleaning up task manager...")
try:
await manager.cleanup()
logger.info("Task manager cleanup completed")
except Exception as e:
logger.error(f"Error cleaning up task manager: {e}")
def get_module_tests(module_path: str) -> List[str]:
"""Get the list of tests in a module."""
logger.info(f"Analyzing module: {module_path}")
with open(module_path, 'r') as file:
content = file.read()
# Pattern to match test functions but exclude fixtures
pattern = r'async\s+def\s+(test_\w+)\s*\('
# Find test functions that are not fixtures (exclude lines with @pytest.fixture)
lines = content.split('\n')
test_functions = []
for i, line in enumerate(lines):
if i > 0 and '@pytest.fixture' in lines[i-1]:
continue # Skip this as it's a fixture, not a test
match = re.search(pattern, line)
if match:
test_functions.append(match.group(1))
logger.info(f"Found {len(test_functions)} tests in {module_path}")
return test_functions
def load_test_module(module_path: str):
"""Load a test module with proper path handling."""
# Convert file path to module path
if module_path.endswith('.py'):
module_path = module_path[:-3] # Remove .py extension
# Convert path separators to module separators
module_name = module_path.replace('/', '.').replace('\\', '.')
# Ensure we use the correct Python path
if not any(p == '.' for p in sys.path):
sys.path.append('.')
logger.info(f"Attempting to import module: {module_name}")
try:
return importlib.import_module(module_name)
except ImportError as e:
logger.error(f"Failed to import test module {module_name}: {e}")
return None
async def run_component_test(module_path: str, test_name: str) -> bool:
"""
Dynamically load and run a component test with proper fixture initialization.
Args:
module_path: Path to the test module
test_name: Name of the test function to run
Returns:
True if test passed, False if it failed
"""
logger.info(f"Running test: {module_path}::{test_name}")
# Import the test module
test_module = load_test_module(module_path)
if not test_module:
return False
# Get the test function
if not hasattr(test_module, test_name):
logger.error(f"Test function {test_name} not found in module {module_name}")
return False
test_func = getattr(test_module, test_name)
# Determine which fixtures the test needs
required_fixtures = inspect.signature(test_func).parameters
logger.info(f"Test requires fixtures: {list(required_fixtures.keys())}")
# Initialize the required fixtures
fixture_values = {}
resources_to_cleanup = []
try:
# Create ServerConfig first since many other fixtures depend on it
if "test_config" in required_fixtures:
logger.info("Setting up test_config fixture")
fixture_values["test_config"] = await create_test_config()
# Create embedder if needed
if "embedder" in required_fixtures:
logger.info("Setting up embedder fixture")
fixture_values["embedder"] = await create_embedder()
# Create test metadata if needed
if "test_metadata" in required_fixtures:
logger.info("Setting up test_metadata fixture")
fixture_values["test_metadata"] = await create_test_metadata()
# Create test code if needed
if "test_code" in required_fixtures:
logger.info("Setting up test_code fixture")
fixture_values["test_code"] = create_test_code()
# Create vector store if needed
if "vector_store" in required_fixtures:
logger.info("Setting up vector_store fixture")
if "test_config" not in fixture_values:
fixture_values["test_config"] = await create_test_config()
if "embedder" not in fixture_values:
fixture_values["embedder"] = await create_embedder()
fixture_values["vector_store"] = await create_vector_store(
fixture_values["test_config"],
fixture_values["embedder"]
)
resources_to_cleanup.append(("vector_store", fixture_values["vector_store"]))
# Create knowledge base if needed
if "knowledge_base" in required_fixtures:
logger.info("Setting up knowledge_base fixture")
if "test_config" not in fixture_values:
fixture_values["test_config"] = await create_test_config()
if "vector_store" not in fixture_values:
if "embedder" not in fixture_values:
fixture_values["embedder"] = await create_embedder()
fixture_values["vector_store"] = await create_vector_store(
fixture_values["test_config"],
fixture_values["embedder"]
)
resources_to_cleanup.append(("vector_store", fixture_values["vector_store"]))
fixture_values["knowledge_base"] = await create_knowledge_base(
fixture_values["test_config"],
fixture_values["vector_store"]
)
resources_to_cleanup.append(("knowledge_base", fixture_values["knowledge_base"]))
# Create task manager if needed
if "task_manager" in required_fixtures:
logger.info("Setting up task_manager fixture")
if "test_config" not in fixture_values:
fixture_values["test_config"] = await create_test_config()
fixture_values["task_manager"] = await create_task_manager(fixture_values["test_config"])
resources_to_cleanup.append(("task_manager", fixture_values["task_manager"]))
# Ensure all required fixtures are initialized
missing_fixtures = set(required_fixtures.keys()) - set(fixture_values.keys())
if missing_fixtures:
logger.error(f"Missing required fixtures: {missing_fixtures}")
return False
# Run the actual test
logger.info(f"Executing test with fixtures: {list(fixture_values.keys())}")
test_kwargs = {name: value for name, value in fixture_values.items() if name in required_fixtures}
# Check if the test function is an async function
if inspect.iscoroutinefunction(test_func):
# For async test functions, await them
logger.info(f"Running async test: {test_name}")
await test_func(**test_kwargs)
else:
# For regular test functions, just call them
logger.info(f"Running synchronous test: {test_name}")
test_func(**test_kwargs)
logger.info(f"Test {test_name} completed successfully")
return True
except Exception as e:
logger.error(f"Test {test_name} failed with error: {e}")
import traceback
logger.error(traceback.format_exc())
return False
finally:
# Clean up resources in reverse order (LIFO)
logger.info("Cleaning up resources...")
for resource_type, resource in reversed(resources_to_cleanup):
try:
if resource_type == "vector_store":
await cleanup_vector_store(resource)
elif resource_type == "knowledge_base":
await cleanup_knowledge_base(resource)
elif resource_type == "task_manager":
await cleanup_task_manager(resource)
except Exception as e:
logger.error(f"Error cleaning up {resource_type}: {e}")
def main():
"""Run a component test with proper async fixture handling."""
if len(sys.argv) < 2:
print("Usage: python component_test_runner.py <module_path> <test_name>")
sys.exit(1)
module_path = sys.argv[1]
# Configure event loop policy for macOS if needed
if sys.platform == 'darwin':
import platform
if int(platform.mac_ver()[0].split('.')[0]) >= 10:
asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
try:
if len(sys.argv) < 3:
# No specific test provided, use module discovery
tests = get_module_tests(module_path)
if not tests:
logger.error(f"No tests found in {module_path}")
sys.exit(1)
# Run all tests in the module
successful_tests = 0
for test_name in tests:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
test_result = loop.run_until_complete(run_component_test(module_path, test_name))
loop.close()
if test_result:
successful_tests += 1
# Report test results
logger.info(f"Test Results: {successful_tests}/{len(tests)} tests passed")
sys.exit(0 if successful_tests == len(tests) else 1)
else:
# Run a specific test
test_name = sys.argv[2]
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(run_component_test(module_path, test_name))
loop.close()
sys.exit(0 if result else 1)
except KeyboardInterrupt:
logger.info("Test execution interrupted")
sys.exit(130) # 130 is the standard exit code for SIGINT
except Exception as e:
logger.error(f"Unhandled exception during test execution: {e}")
import traceback
logger.error(traceback.format_exc())
sys.exit(1)
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/trajectories/tosinakinosho/default__openrouter/anthropic/claude-3.5-sonnet-20240620:beta__t-0.00__p-1.00__c-3.00___03565e/03565e/config.yaml:
--------------------------------------------------------------------------------
```yaml
'{"env":{"deployment":{"image":"python:3.11","port":null,"docker_args":[],"startup_timeout":180.0,"pull":"missing","remove_images":false,"python_standalone_dir":"/root","platform":null,"type":"docker"},"repo":{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight","base_commit":"HEAD","type":"local"},"post_startup_commands":[],"post_startup_command_timeout":500,"name":"main"},"agent":{"name":"main","templates":{"system_template":"SETTING:
You are an autonomous programmer, and you''re working directly in the command line
with a special interface.\n\nThe special interface consists of a file editor that
shows you {{WINDOW}} lines of a file at a time.\nIn addition to typical bash commands,
you can also use specific commands to help you navigate and edit files.\nTo call
a command, you need to invoke it with a function call/tool call.\n\nPlease note
that THE EDIT COMMAND REQUIRES PROPER INDENTATION.\n\nFor example, if you are looking
at this file:\n\ndef fct():\n print(\"Hello world\")\n\nand you want to edit
the file to read:\n\ndef fct():\n print(\"Hello\")\n print(\"world\")\n\nyou
search string should be `Hello world` and your replace string should be `\"Hello\"\\n print(\"world\")`\n(note
the extra spaces before the print statement!).\n\nYou could also get the same result
by search for ` print(\"Hello world\")` and replace with ` print(\"Hello\")\\n print(\"world\")`.\n\nRESPONSE
FORMAT:\nYour shell prompt is formatted as follows:\n(Open file: <path>)\n(Current
directory: <cwd>)\nbash-$\n\nFirst, you should _always_ include a general thought
about what you''re going to do next.\nThen, for every response, you must include
exactly _ONE_ tool call/function call.\n\nRemember, you should always include a
_SINGLE_ tool call/function call and then wait for a response from the shell before
continuing with more discussion and commands. Everything you include in the DISCUSSION
section will be saved for future reference.\nIf you''d like to issue two commands
at once, PLEASE DO NOT DO THAT! Please instead first submit just the first tool
call, and then after receiving a response you''ll be able to issue the second .\nNote
that the environment does NOT support interactive session commands (e.g. python,
vim), so please do not invoke them.","instance_template":"We''re currently solving
the following issue within our repository. Here''s the issue text:\nISSUE:\n{{problem_statement}}\n\nINSTRUCTIONS:\nNow,
you''re going to solve this issue on your own. Your terminal session has started
and you''re in the repository''s root directory. You can use any bash commands or
the special interface to help you. Edit all the files you need to and run any checks
or tests that you want.\nRemember, YOU SHOULD ALWAYS INCLUDE EXACTLY ONE TOOL CALL/FUNCTION
CALL PER RESPONSE.\nWhen you''re satisfied with all of the changes you''ve made,
you can submit your changes to the code base by simply running the submit command.\nNote
however that you cannot use any interactive session commands (e.g. python, vim)
in this environment, but you can write scripts and run them. E.g. you can write
a python script and then run it with the python command.\n\nNOTE ABOUT THE EDIT
COMMAND: Indentation really matters! When editing a file, make sure to insert appropriate
indentation before each line!\n\nGENERAL IMPORTANT TIPS:\n\n1. If you run a command
and it doesn''t work, try running a different command. A command that did not work
once will not work the second time unless you modify it!\n\n2. If you open a file
and need to get to an area around a specific line that is not in the first 100 lines,
say line 583, don''t just use the scroll_down command multiple times. Instead, use
the goto 583 command. It''s much quicker.\n\n3. If the bug reproduction script requires
inputting/reading a specific file, such as buggy-input.png, and you''d like to understand
how to input that file, conduct a search in the existing repo code, to see whether
someone else has already done that. Do this by running the command: find_file \"buggy-input.png\"
If that doesn''t work, use the linux ''find'' command.\n\n4. Always make sure to
look at the currently open file and the current working directory (which appears
right after the currently open file). The currently open file might be in a different
directory than the working directory! Note that some commands, such as ''create'',
open files, so they might change the current open file.\n\n5. When editing files,
it is easy to accidentally to write code with incorrect indentation or make other
mistakes. Always check the code after you issue an edit to make sure that it reflects
what you wanted to accomplish. If it didn''t, issue another command to fix it.\n\n6.
When editing files, first explain the code you want to edit and why it is causing
the problem. Then explain the edit you want to make and how it fixes the problem.
Explain how the edit does not break existing functionality.\n\n7. Do not try to
install any packages with `pip`, `conda`, or any other way. This will usually not
work. If the environment is not set up correctly, try to fix the issue without executing
python code or running any tests that require the package installed.\n\nSTRATEGY:\n\n1.
Always start by trying to replicate the bug that the issues discusses.\n If the
issue includes code for reproducing the bug, we recommend that you re-implement
that in your environment, and run it to make sure you can reproduce the bug.\n Then
start trying to fix it.\n\n If the bug reproduction script does not print anything
when it successfully runs, we recommend adding a print(\"Script completed successfully,
no errors.\") command at the end of the file,\n so that you can be sure that the
script indeed ran fine all the way through.\n\n2. Locate relevant code using the
find and search commands. `open` the file you want to edit.\n\n3. Use the `edit`
command to perform edits.\n\n4. When you think you''ve fixed the bug, re-run the
bug reproduction script to make sure that the bug has indeed been fixed.\n\n5. Create
additional tests to verify the fix in a style similar to the existing reproduction
script. In particular, make sure to test edge cases.\n If you find any issues,
go back to the file you edited and perform further edits.\n\n(Open file: {{open_file}})\n(Current
directory: {{working_dir}})\nbash-$","next_step_template":"{{observation}}\n(Open
file: {{open_file}})\n(Current directory: {{working_dir}})\nbash-$","next_step_truncated_observation_template":"Observation:
{{observation}}<response clipped><NOTE>Observations should not exceeded {{max_observation_length}}
characters. {{elided_chars}} characters were elided. Please try a different command
that produces less output or use head/tail/grep/redirect the output to a file. Do
not use interactive pagers.</NOTE>","max_observation_length":100000,"next_step_no_output_template":"Your
command ran successfully and did not produce any output.\n(Open file: {{open_file}})\n(Current
directory: {{working_dir}})\nbash-$","strategy_template":null,"demonstration_template":"Here
is a demonstration of how to correctly accomplish this task.\nIt is included to
show you how to correctly use the interface.\nYou do not need to follow exactly
what is done in the demonstration.\n--- DEMONSTRATION ---\n{{demonstration}}\n---
END OF DEMONSTRATION ---\n","demonstrations":["/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/trajectories/demonstrations/replay__marshmallow-code__marshmallow-1867__function_calling_replace__install-1/marshmallow-code__marshmallow-1867.traj"],"put_demos_in_history":true,"shell_check_error_template":"Your
bash command contained syntax errors and was NOT executed. Please fix the syntax
errors and try again. This can be the result of not adhering to the syntax for multi-line
commands. Here is the output of `bash -n`:\n{{bash_stdout}}\n{{bash_stderr}}","command_cancelled_timeout_template":"The
command ''{{command}}'' was cancelled because it took more than {{timeout}} seconds.
Please try a different command that completes more quickly."},"tools":{"filter":{"blocklist_error_template":"Operation
''{{action}}'' is not supported by this environment.","blocklist":["vim","vi","emacs","nano","nohup","gdb","less","tail
-f","python -m venv","make"],"blocklist_standalone":["python","python3","ipython","bash","sh","/bin/bash","/bin/sh","nohup","vi","vim","emacs","nano","su"],"block_unless_regex":{"radare2":"\\b(?:radare2)\\b.*\\s+-c\\s+.*","r2":"\\b(?:radare2)\\b.*\\s+-c\\s+.*"}},"bundles":[{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/registry","hidden_tools":[]},{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/defaults","hidden_tools":[]},{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/search","hidden_tools":[]},{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/edit_replace","hidden_tools":[]},{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/submit","hidden_tools":[]}],"env_variables":{"WINDOW":100,"OVERLAP":2},"registry_variables":{},"submit_command":"submit","parse_function":{"error_message":"Your
output was not formatted correctly. You must always include one discussion and one
command as part of your response. Make sure you do not have multiple discussion/command
tags.\nPlease make sure your output precisely matches the following format:\nDISCUSSION\nDiscuss
here with yourself about what your planning and what you''re going to do in this
step.\n\n```\ncommand(s) that you''re going to run\n```\n","type":"thought_action"},"enable_bash_tool":true,"format_error_template":"Your
output was not formatted correctly. You must always include one discussion and one
command as part of your response. Make sure you do not have multiple discussion/command
tags.\nPlease make sure your output precisely matches the following format:\nDISCUSSION\nDiscuss
here with yourself about what your planning and what you''re going to do in this
step.\n\n```\ncommand(s) that you''re going to run\n```\n","command_docs":"bash:\n docstring:
runs the given command directly in bash\n signature: <command>\n arguments:\n -
command (string) [required]: The bash command to execute.\n\ngoto:\n docstring:
moves the window to show <line_number>\n signature: goto <line_number>\n arguments:\n -
line_number (integer) [required]: the line number to move the window to\n\nopen:\n docstring:
opens the file at the given path in the editor. If line_number is provided, the
window will be move to include that line\n signature: open \"<path>\" [<line_number>]\n arguments:\n -
path (string) [required]: the path to the file to open\n - line_number (integer)
[optional]: the line number to move the window to (if not provided, the window will
start at the top of the file)\n\ncreate:\n docstring: creates and opens a new file
with the given name\n signature: create <filename>\n arguments:\n - filename
(string) [required]: the name of the file to create\n\nscroll_up:\n docstring:
moves the window up 100 lines\n signature: scroll_up\n\nscroll_down:\n docstring:
moves the window down 100 lines\n signature: scroll_down\n\nfind_file:\n docstring:
finds all files with the given name or pattern in dir. If dir is not provided, searches
in the current directory\n signature: find_file <file_name> [<dir>]\n arguments:\n -
file_name (string) [required]: the name of the file or pattern to search for. supports
shell-style wildcards (e.g. *.py)\n - dir (string) [optional]: the directory
to search in (if not provided, searches in the current directory)\n\nsearch_dir:\n docstring:
searches for search_term in all files in dir. If dir is not provided, searches in
the current directory\n signature: search_dir <search_term> [<dir>]\n arguments:\n -
search_term (string) [required]: the term to search for\n - dir (string) [optional]:
the directory to search in (if not provided, searches in the current directory)\n\nsearch_file:\n docstring:
searches for search_term in file. If file is not provided, searches in the current
open file\n signature: search_file <search_term> [<file>]\n arguments:\n -
search_term (string) [required]: the term to search for\n - file (string) [optional]:
the file to search in (if not provided, searches in the current open file)\n\nedit:\n docstring:
Replace first occurrence of <search> with <replace> in the currently displayed lines.
If replace-all is True , replace all occurrences of <search> with <replace>.\nFor
example, if you are looking at this file:\ndef fct():\n print(\"Hello world\")\n\nand
you want to edit the file to read:\ndef fct():\n print(\"Hello\")\n print(\"world\")\n\nyou
can search for `Hello world` and replace with `\"Hello\"\\n print(\"world\")`
(note the extra spaces before the print statement!).\nTips:\n1. Always include proper
whitespace/indentation 2. When you are adding an if/with/try statement, you need
to INDENT the block that follows, so make sure to include it in both your search
and replace strings! 3. If you are wrapping code in a try statement, make sure to
also add an ''except'' or ''finally'' block.\nBefore every edit, please\n1. Explain
the code you want to edit and why it is causing the problem 2. Explain the edit
you want to make and how it fixes the problem 3. Explain how the edit does not break
existing functionality\n\n signature: edit <search> <replace> [<replace-all>]\n\n arguments:\n -
search (string) [required]: the text to search for (make sure to include proper
whitespace if needed)\n - replace (string) [required]: the text to replace the
search with (make sure to include proper whitespace if needed)\n - replace-all
(boolean) [optional]: replace all occurrences rather than the first occurrence within
the displayed lines\n\ninsert:\n docstring: Insert <text> at the end of the currently
opened file or after <line> if specified.\n\n signature: insert <text> [<line>]\n\n arguments:\n -
text (string) [required]: the text to insert\n - line (integer) [optional]: the
line number to insert the text as new lines after\n\nsubmit:\n docstring: submits
the current file\n signature: submit\n\n","multi_line_command_endings":{},"submit_command_end_name":null,"reset_commands":[],"execution_timeout":30,"install_timeout":300,"total_execution_timeout":1800,"max_consecutive_execution_timeouts":3},"history_processors":[{"n":5,"polling":1,"always_remove_output_for_tags":["remove_output"],"always_keep_output_for_tags":["keep_output"],"type":"last_n_observations"}],"model":{"name":"openrouter/anthropic/claude-3.5-sonnet-20240620:beta","per_instance_cost_limit":3.0,"total_cost_limit":0.0,"per_instance_call_limit":0,"temperature":0.0,"top_p":1.0,"api_base":null,"api_version":null,"api_key":null,"stop":[],"completion_kwargs":{},"convert_system_to_user":false,"retry":{"retries":20,"min_wait":10.0,"max_wait":120.0},"delay":0.0,"fallbacks":[],"choose_api_key_by_thread":true,"max_input_tokens":null,"max_output_tokens":null},"max_requeries":3,"action_sampler":null,"type":"default"},"problem_statement":{"text":"#
Debug MCP Codebase Insight Tests","extra_fields":{},"type":"text","id":"03565e"},"output_dir":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/trajectories/tosinakinosho/default__openrouter/anthropic/claude-3.5-sonnet-20240620:beta__t-0.00__p-1.00__c-3.00___03565e","actions":{"open_pr":false,"pr_config":{"skip_if_commits_reference_issue":true},"apply_patch_locally":false},"env_var_path":null}'
```
--------------------------------------------------------------------------------
/trajectories/tosinakinosho/default__claude-3-5-sonnet-20240620__t-0.00__p-1.00__c-3.00___03565e/03565e/config.yaml:
--------------------------------------------------------------------------------
```yaml
'{"env":{"deployment":{"image":"python:3.11","port":null,"docker_args":[],"startup_timeout":180.0,"pull":"missing","remove_images":false,"python_standalone_dir":"/root","platform":null,"type":"docker"},"repo":null,"post_startup_commands":[],"post_startup_command_timeout":500,"name":"main"},"agent":{"name":"main","templates":{"system_template":"SETTING:
You are an autonomous programmer, and you''re working directly in the command line
with a special interface.\n\nThe special interface consists of a file editor that
shows you {{WINDOW}} lines of a file at a time.\nIn addition to typical bash commands,
you can also use specific commands to help you navigate and edit files.\nTo call
a command, you need to invoke it with a function call/tool call.\n\nPlease note
that THE EDIT COMMAND REQUIRES PROPER INDENTATION.\n\nFor example, if you are looking
at this file:\n\ndef fct():\n print(\"Hello world\")\n\nand you want to edit
the file to read:\n\ndef fct():\n print(\"Hello\")\n print(\"world\")\n\nyou
search string should be `Hello world` and your replace string should be `\"Hello\"\\n print(\"world\")`\n(note
the extra spaces before the print statement!).\n\nYou could also get the same result
by search for ` print(\"Hello world\")` and replace with ` print(\"Hello\")\\n print(\"world\")`.\n\nRESPONSE
FORMAT:\nYour shell prompt is formatted as follows:\n(Open file: <path>)\n(Current
directory: <cwd>)\nbash-$\n\nFirst, you should _always_ include a general thought
about what you''re going to do next.\nThen, for every response, you must include
exactly _ONE_ tool call/function call.\n\nRemember, you should always include a
_SINGLE_ tool call/function call and then wait for a response from the shell before
continuing with more discussion and commands. Everything you include in the DISCUSSION
section will be saved for future reference.\nIf you''d like to issue two commands
at once, PLEASE DO NOT DO THAT! Please instead first submit just the first tool
call, and then after receiving a response you''ll be able to issue the second .\nNote
that the environment does NOT support interactive session commands (e.g. python,
vim), so please do not invoke them.","instance_template":"We''re currently solving
the following issue within our repository. Here''s the issue text:\nISSUE:\n{{problem_statement}}\n\nINSTRUCTIONS:\nNow,
you''re going to solve this issue on your own. Your terminal session has started
and you''re in the repository''s root directory. You can use any bash commands or
the special interface to help you. Edit all the files you need to and run any checks
or tests that you want.\nRemember, YOU SHOULD ALWAYS INCLUDE EXACTLY ONE TOOL CALL/FUNCTION
CALL PER RESPONSE.\nWhen you''re satisfied with all of the changes you''ve made,
you can submit your changes to the code base by simply running the submit command.\nNote
however that you cannot use any interactive session commands (e.g. python, vim)
in this environment, but you can write scripts and run them. E.g. you can write
a python script and then run it with the python command.\n\nNOTE ABOUT THE EDIT
COMMAND: Indentation really matters! When editing a file, make sure to insert appropriate
indentation before each line!\n\nGENERAL IMPORTANT TIPS:\n\n1. If you run a command
and it doesn''t work, try running a different command. A command that did not work
once will not work the second time unless you modify it!\n\n2. If you open a file
and need to get to an area around a specific line that is not in the first 100 lines,
say line 583, don''t just use the scroll_down command multiple times. Instead, use
the goto 583 command. It''s much quicker.\n\n3. If the bug reproduction script requires
inputting/reading a specific file, such as buggy-input.png, and you''d like to understand
how to input that file, conduct a search in the existing repo code, to see whether
someone else has already done that. Do this by running the command: find_file \"buggy-input.png\"
If that doesn''t work, use the linux ''find'' command.\n\n4. Always make sure to
look at the currently open file and the current working directory (which appears
right after the currently open file). The currently open file might be in a different
directory than the working directory! Note that some commands, such as ''create'',
open files, so they might change the current open file.\n\n5. When editing files,
it is easy to accidentally to write code with incorrect indentation or make other
mistakes. Always check the code after you issue an edit to make sure that it reflects
what you wanted to accomplish. If it didn''t, issue another command to fix it.\n\n6.
When editing files, first explain the code you want to edit and why it is causing
the problem. Then explain the edit you want to make and how it fixes the problem.
Explain how the edit does not break existing functionality.\n\n7. Do not try to
install any packages with `pip`, `conda`, or any other way. This will usually not
work. If the environment is not set up correctly, try to fix the issue without executing
python code or running any tests that require the package installed.\n\nSTRATEGY:\n\n1.
Always start by trying to replicate the bug that the issues discusses.\n If the
issue includes code for reproducing the bug, we recommend that you re-implement
that in your environment, and run it to make sure you can reproduce the bug.\n Then
start trying to fix it.\n\n If the bug reproduction script does not print anything
when it successfully runs, we recommend adding a print(\"Script completed successfully,
no errors.\") command at the end of the file,\n so that you can be sure that the
script indeed ran fine all the way through.\n\n2. Locate relevant code using the
find and search commands. `open` the file you want to edit.\n\n3. Use the `edit`
command to perform edits.\n\n4. When you think you''ve fixed the bug, re-run the
bug reproduction script to make sure that the bug has indeed been fixed.\n\n5. Create
additional tests to verify the fix in a style similar to the existing reproduction
script. In particular, make sure to test edge cases.\n If you find any issues,
go back to the file you edited and perform further edits.\n\n(Open file: {{open_file}})\n(Current
directory: {{working_dir}})\nbash-$","next_step_template":"{{observation}}\n(Open
file: {{open_file}})\n(Current directory: {{working_dir}})\nbash-$","next_step_truncated_observation_template":"Observation:
{{observation}}<response clipped><NOTE>Observations should not exceeded {{max_observation_length}}
characters. {{elided_chars}} characters were elided. Please try a different command
that produces less output or use head/tail/grep/redirect the output to a file. Do
not use interactive pagers.</NOTE>","max_observation_length":100000,"next_step_no_output_template":"Your
command ran successfully and did not produce any output.\n(Open file: {{open_file}})\n(Current
directory: {{working_dir}})\nbash-$","strategy_template":null,"demonstration_template":"Here
is a demonstration of how to correctly accomplish this task.\nIt is included to
show you how to correctly use the interface.\nYou do not need to follow exactly
what is done in the demonstration.\n--- DEMONSTRATION ---\n{{demonstration}}\n---
END OF DEMONSTRATION ---\n","demonstrations":["/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/trajectories/demonstrations/replay__marshmallow-code__marshmallow-1867__function_calling_replace__install-1/marshmallow-code__marshmallow-1867.traj"],"put_demos_in_history":true,"shell_check_error_template":"Your
bash command contained syntax errors and was NOT executed. Please fix the syntax
errors and try again. This can be the result of not adhering to the syntax for multi-line
commands. Here is the output of `bash -n`:\n{{bash_stdout}}\n{{bash_stderr}}","command_cancelled_timeout_template":"The
command ''{{command}}'' was cancelled because it took more than {{timeout}} seconds.
Please try a different command that completes more quickly."},"tools":{"filter":{"blocklist_error_template":"Operation
''{{action}}'' is not supported by this environment.","blocklist":["vim","vi","emacs","nano","nohup","gdb","less","tail
-f","python -m venv","make"],"blocklist_standalone":["python","python3","ipython","bash","sh","/bin/bash","/bin/sh","nohup","vi","vim","emacs","nano","su"],"block_unless_regex":{"radare2":"\\b(?:radare2)\\b.*\\s+-c\\s+.*","r2":"\\b(?:radare2)\\b.*\\s+-c\\s+.*"}},"bundles":[{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/registry","hidden_tools":[]},{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/defaults","hidden_tools":[]},{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/search","hidden_tools":[]},{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/edit_replace","hidden_tools":[]},{"path":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/SWE-agent/tools/submit","hidden_tools":[]}],"env_variables":{"WINDOW":100,"OVERLAP":2},"registry_variables":{},"submit_command":"submit","parse_function":{"error_message":"{%-
if error_code == \"missing\" -%}\nYour last output did not use any tool calls!\nPlease
make sure your output includes exactly _ONE_ function call!\nYou must invoke the
function directly using the function call format.\nYou cannot invoke commands with
```, you have to use the function call format.\nIf you think you have already resolved
the issue, please submit your changes by running the `submit` command.\nIf you think
you cannot solve the problem, please run `exit_forfeit` (if available) or `submit`.\nElse,
please continue with a new tool call!\n{%- elif error_code == \"multiple\" -%}\nYour
last output included multiple tool calls!\nPlease make sure your output includes
a thought and exactly _ONE_ function call.\n{%- elif error_code == \"unexpected_arg\"
-%}\nYour action could not be parsed properly: {{exception_message}}.\nMake sure
your function call doesn''t include any extra arguments that are not in the allowed
arguments, and only use the allowed commands.\n{%- else -%}\nYour action could not
be parsed properly: {{exception_message}}.\n{% endif %}\n","type":"function_calling"},"enable_bash_tool":true,"format_error_template":"{%-
if error_code == \"missing\" -%}\nYour last output did not use any tool calls!\nPlease
make sure your output includes exactly _ONE_ function call!\nYou must invoke the
function directly using the function call format.\nYou cannot invoke commands with
```, you have to use the function call format.\nIf you think you have already resolved
the issue, please submit your changes by running the `submit` command.\nIf you think
you cannot solve the problem, please run `exit_forfeit` (if available) or `submit`.\nElse,
please continue with a new tool call!\n{%- elif error_code == \"multiple\" -%}\nYour
last output included multiple tool calls!\nPlease make sure your output includes
a thought and exactly _ONE_ function call.\n{%- elif error_code == \"unexpected_arg\"
-%}\nYour action could not be parsed properly: {{exception_message}}.\nMake sure
your function call doesn''t include any extra arguments that are not in the allowed
arguments, and only use the allowed commands.\n{%- else -%}\nYour action could not
be parsed properly: {{exception_message}}.\n{% endif %}\n","command_docs":"bash:\n docstring:
runs the given command directly in bash\n signature: <command>\n arguments:\n -
command (string) [required]: The bash command to execute.\n\ngoto:\n docstring:
moves the window to show <line_number>\n signature: goto <line_number>\n arguments:\n -
line_number (integer) [required]: the line number to move the window to\n\nopen:\n docstring:
opens the file at the given path in the editor. If line_number is provided, the
window will be move to include that line\n signature: open \"<path>\" [<line_number>]\n arguments:\n -
path (string) [required]: the path to the file to open\n - line_number (integer)
[optional]: the line number to move the window to (if not provided, the window will
start at the top of the file)\n\ncreate:\n docstring: creates and opens a new file
with the given name\n signature: create <filename>\n arguments:\n - filename
(string) [required]: the name of the file to create\n\nscroll_up:\n docstring:
moves the window up 100 lines\n signature: scroll_up\n\nscroll_down:\n docstring:
moves the window down 100 lines\n signature: scroll_down\n\nfind_file:\n docstring:
finds all files with the given name or pattern in dir. If dir is not provided, searches
in the current directory\n signature: find_file <file_name> [<dir>]\n arguments:\n -
file_name (string) [required]: the name of the file or pattern to search for. supports
shell-style wildcards (e.g. *.py)\n - dir (string) [optional]: the directory
to search in (if not provided, searches in the current directory)\n\nsearch_dir:\n docstring:
searches for search_term in all files in dir. If dir is not provided, searches in
the current directory\n signature: search_dir <search_term> [<dir>]\n arguments:\n -
search_term (string) [required]: the term to search for\n - dir (string) [optional]:
the directory to search in (if not provided, searches in the current directory)\n\nsearch_file:\n docstring:
searches for search_term in file. If file is not provided, searches in the current
open file\n signature: search_file <search_term> [<file>]\n arguments:\n -
search_term (string) [required]: the term to search for\n - file (string) [optional]:
the file to search in (if not provided, searches in the current open file)\n\nedit:\n docstring:
Replace first occurrence of <search> with <replace> in the currently displayed lines.
If replace-all is True , replace all occurrences of <search> with <replace>.\nFor
example, if you are looking at this file:\ndef fct():\n print(\"Hello world\")\n\nand
you want to edit the file to read:\ndef fct():\n print(\"Hello\")\n print(\"world\")\n\nyou
can search for `Hello world` and replace with `\"Hello\"\\n print(\"world\")`
(note the extra spaces before the print statement!).\nTips:\n1. Always include proper
whitespace/indentation 2. When you are adding an if/with/try statement, you need
to INDENT the block that follows, so make sure to include it in both your search
and replace strings! 3. If you are wrapping code in a try statement, make sure to
also add an ''except'' or ''finally'' block.\nBefore every edit, please\n1. Explain
the code you want to edit and why it is causing the problem 2. Explain the edit
you want to make and how it fixes the problem 3. Explain how the edit does not break
existing functionality\n\n signature: edit <search> <replace> [<replace-all>]\n\n arguments:\n -
search (string) [required]: the text to search for (make sure to include proper
whitespace if needed)\n - replace (string) [required]: the text to replace the
search with (make sure to include proper whitespace if needed)\n - replace-all
(boolean) [optional]: replace all occurrences rather than the first occurrence within
the displayed lines\n\ninsert:\n docstring: Insert <text> at the end of the currently
opened file or after <line> if specified.\n\n signature: insert <text> [<line>]\n\n arguments:\n -
text (string) [required]: the text to insert\n - line (integer) [optional]: the
line number to insert the text as new lines after\n\nsubmit:\n docstring: submits
the current file\n signature: submit\n\n","multi_line_command_endings":{},"submit_command_end_name":null,"reset_commands":[],"execution_timeout":30,"install_timeout":300,"total_execution_timeout":1800,"max_consecutive_execution_timeouts":3},"history_processors":[{"n":5,"polling":1,"always_remove_output_for_tags":["remove_output"],"always_keep_output_for_tags":["keep_output"],"type":"last_n_observations"}],"model":{"name":"claude-3-5-sonnet-20240620","per_instance_cost_limit":3.0,"total_cost_limit":0.0,"per_instance_call_limit":0,"temperature":0.0,"top_p":1.0,"api_base":null,"api_version":null,"api_key":null,"stop":[],"completion_kwargs":{},"convert_system_to_user":false,"retry":{"retries":20,"min_wait":10.0,"max_wait":120.0},"delay":0.0,"fallbacks":[],"choose_api_key_by_thread":true,"max_input_tokens":null,"max_output_tokens":null},"max_requeries":3,"action_sampler":null,"type":"default"},"problem_statement":{"text":"#
Debug MCP Codebase Insight Tests","extra_fields":{},"type":"text","id":"03565e"},"output_dir":"/Users/tosinakinosho/workspaces/mcp-codebase-insight/trajectories/tosinakinosho/default__claude-3-5-sonnet-20240620__t-0.00__p-1.00__c-3.00___03565e","actions":{"open_pr":false,"pr_config":{"skip_if_commits_reference_issue":true},"apply_patch_locally":false},"env_var_path":null}'
```
--------------------------------------------------------------------------------
/src/mcp_codebase_insight/core/sse.py:
--------------------------------------------------------------------------------
```python
"""Server-Sent Events (SSE) transport implementation for MCP."""
import asyncio
import logging
import json
from typing import Any, Callable, Dict, List, Optional, Tuple
from datetime import datetime
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from starlette.requests import Request
from starlette.responses import Response, JSONResponse, RedirectResponse, StreamingResponse
import uuid
from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
from starlette.middleware.cors import CORSMiddleware
from mcp.server.fastmcp import FastMCP
from mcp.server.sse import SseServerTransport
from ..utils.logger import get_logger
logger = get_logger(__name__)
async def send_heartbeats(queue: asyncio.Queue, interval: int = 30):
"""Send periodic heartbeat messages to keep the connection alive.
Args:
queue: The queue to send heartbeats to
interval: Time between heartbeats in seconds
"""
while True:
try:
await queue.put({"type": "heartbeat", "timestamp": datetime.utcnow().isoformat()})
await asyncio.sleep(interval)
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Error sending heartbeat: {e}")
await asyncio.sleep(1) # Brief pause before retrying
class CodebaseInsightSseTransport(SseServerTransport):
"""Custom SSE transport implementation for Codebase Insight."""
def __init__(self, endpoint: str):
"""Initialize the SSE transport.
Args:
endpoint: The endpoint path for SSE connections
"""
super().__init__(endpoint)
self.connections = {}
self.message_queue = asyncio.Queue()
logger.info(f"Initializing SSE transport with endpoint: {endpoint}")
async def handle_sse(self, request: Request) -> StreamingResponse:
"""Handle incoming SSE connection requests.
Args:
request: The incoming HTTP request
Returns:
StreamingResponse for the SSE connection
"""
connection_id = str(uuid.uuid4())
queue = asyncio.Queue()
self.connections[connection_id] = queue
logger.info(f"New SSE connection established: {connection_id}")
logger.debug(f"Request headers: {dict(request.headers)}")
logger.debug(f"Active connections: {len(self.connections)}")
async def event_generator():
try:
logger.debug(f"Starting event generator for connection {connection_id}")
heartbeat_task = asyncio.create_task(send_heartbeats(queue))
logger.debug(f"Heartbeat task started for connection {connection_id}")
while True:
try:
message = await queue.get()
logger.debug(f"Connection {connection_id} received message: {message}")
if isinstance(message, dict):
data = json.dumps(message)
else:
data = str(message)
yield f"data: {data}\n\n"
logger.debug(f"Sent message to connection {connection_id}")
except asyncio.CancelledError:
logger.info(f"Event generator cancelled for connection {connection_id}")
break
except Exception as e:
logger.error(f"Error in event generator for connection {connection_id}: {e}")
break
finally:
heartbeat_task.cancel()
try:
await heartbeat_task
except asyncio.CancelledError:
pass
if connection_id in self.connections:
del self.connections[connection_id]
logger.info(f"Event generator cleaned up for connection {connection_id}")
logger.debug(f"Remaining active connections: {len(self.connections)}")
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
"Access-Control-Allow-Origin": "*", # Allow CORS
"Access-Control-Allow-Headers": "Content-Type",
"Access-Control-Allow-Methods": "GET, POST"
}
)
async def handle_message(self, request: Request) -> Response:
"""Handle incoming messages to be broadcast over SSE.
Args:
request: The incoming HTTP request with the message
Returns:
HTTP response indicating message handling status
"""
try:
message = await request.json()
# Broadcast to all connections
for queue in self.connections.values():
await queue.put(message)
return JSONResponse({"status": "message sent"})
except Exception as e:
logger.error(f"Error handling message: {e}")
return JSONResponse(
{"error": str(e)},
status_code=500
)
async def send(self, message: Any) -> None:
"""Send a message to all connected clients.
Args:
message: The message to send
"""
# Put message in queue for all connections
for queue in self.connections.values():
await queue.put(message)
async def broadcast(self, message: Any) -> None:
"""Broadcast a message to all connected clients.
Args:
message: The message to broadcast
"""
await self.send(message)
async def connect(self) -> Tuple[MemoryObjectReceiveStream, MemoryObjectSendStream]:
"""Create a new SSE connection.
Returns:
Tuple of receive and send streams for the connection
"""
# Create memory object streams for this connection
receive_stream = MemoryObjectReceiveStream()
send_stream = MemoryObjectSendStream()
# Store the connection
connection_id = str(uuid.uuid4())
self.connections[connection_id] = send_stream
return receive_stream, send_stream
async def disconnect(self, connection_id: str) -> None:
"""Disconnect a client.
Args:
connection_id: The ID of the connection to disconnect
"""
if connection_id in self.connections:
del self.connections[connection_id]
logger.info(f"Disconnected client: {connection_id}")
async def verify_routes(app: Starlette) -> Dict[str, List[str]]:
"""Verify and log all registered routes in the application.
Args:
app: The Starlette application to verify
Returns:
Dictionary mapping route paths to their methods
"""
routes = {}
for route in app.routes:
if isinstance(route, Mount):
logger.info(f"Mount point: {route.path}")
# Recursively verify mounted routes
mounted_routes = await verify_routes(route.app)
for path, methods in mounted_routes.items():
full_path = f"{route.path}{path}"
routes[full_path] = methods
else:
routes[route.path] = route.methods
logger.info(f"Route: {route.path}, methods: {route.methods}")
return routes
def create_sse_server(mcp_server: Optional[FastMCP] = None) -> Starlette:
"""Create an SSE server instance.
Args:
mcp_server: Optional FastMCP instance to use. If not provided, a new one will be created.
Returns:
Starlette application configured for SSE
"""
app = Starlette(debug=True) # Enable debug mode for better error reporting
# Create SSE transport
transport = CodebaseInsightSseTransport("/sse")
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allow all origins
allow_credentials=True,
allow_methods=["GET", "POST", "OPTIONS"],
allow_headers=["*"],
expose_headers=["*"]
)
async def health_check(request: Request) -> JSONResponse:
"""Health check endpoint."""
return JSONResponse({
"status": "ok",
"timestamp": datetime.utcnow().isoformat(),
"connections": len(transport.connections)
})
# Add routes
app.add_route("/health", health_check, methods=["GET"])
app.add_route("/sse", transport.handle_sse, methods=["GET"])
app.add_route("/message", transport.handle_message, methods=["POST"])
logger.info("Created SSE server with routes:")
asyncio.create_task(verify_routes(app))
return app
class MCP_CodebaseInsightServer:
"""MCP server implementation for Codebase Insight.
This class manages the Model Context Protocol server, connecting it to
the Codebase Insight's core components and exposing them as MCP tools.
"""
def __init__(self, server_state):
"""Initialize the MCP server with access to the application state.
Args:
server_state: The global server state providing access to components
"""
self.state = server_state
self.mcp_server = FastMCP(name="MCP-Codebase-Insight")
self.tools_registered = False
self._starlette_app = None # Cache the Starlette app
logger.info("MCP Codebase Insight server initialized")
async def cleanup(self):
"""Clean up resources used by the MCP server.
This method ensures proper shutdown of the MCP server and
releases any resources it might be holding.
"""
logger.info("Cleaning up MCP server resources")
# If the MCP server has a shutdown or cleanup method, call it here
# For now, just log the cleanup attempt
self.tools_registered = False
self._starlette_app = None
logger.info("MCP server cleanup completed")
def is_initialized(self) -> bool:
"""Check if the MCP server is properly initialized.
Returns:
True if the server is initialized and ready to use, False otherwise
"""
return self.tools_registered and self._starlette_app is not None
def register_tools(self):
"""Register all available tools with the MCP server.
This connects the MCP protocol to the Codebase Insight core components,
exposing their functionality through the MCP interface.
"""
if self.tools_registered:
logger.debug("Tools already registered, skipping")
return
logger.info("Registering tools with MCP server")
# Check if critical dependencies are available
critical_dependencies = ["vector_store", "knowledge_base", "task_manager", "adr_manager"]
missing_dependencies = []
for dependency in critical_dependencies:
if not self.state.get_component(dependency):
missing_dependencies.append(dependency)
if missing_dependencies:
logger.warning(f"Some critical dependencies are not available: {', '.join(missing_dependencies)}")
logger.warning("Tools requiring these dependencies will not be registered")
# Don't fail registration completely - continue with available tools
# Register available tools
try:
self._register_vector_search()
self._register_knowledge()
self._register_adr()
self._register_task()
# Mark tools as registered even if some failed
self.tools_registered = True
logger.info("MCP tools registration completed")
except Exception as e:
logger.error(f"Error registering MCP tools: {e}", exc_info=True)
# Don't mark as registered if there was an error
def _register_vector_search(self):
"""Register the vector search tool with the MCP server."""
vector_store = self.state.get_component("vector_store")
if not vector_store:
logger.warning("Vector store component not available, skipping tool registration")
return
# Verify that the vector store is properly initialized
if not hasattr(vector_store, 'search') or not callable(getattr(vector_store, 'search')):
logger.warning("Vector store component does not have a search method, skipping tool registration")
return
async def vector_search(query: str, limit: int = 5, threshold: float = 0.7,
file_type: Optional[str] = None, path_pattern: Optional[str] = None):
"""Search for code snippets semantically similar to the query text."""
logger.debug(f"MCP vector search request: {query=}, {limit=}, {threshold=}")
# Prepare filters if provided
filter_conditions = {}
if file_type:
filter_conditions["file_type"] = {"$eq": file_type}
if path_pattern:
filter_conditions["path"] = {"$like": path_pattern}
results = await vector_store.search(
text=query,
filter_conditions=filter_conditions if filter_conditions else None,
limit=limit
)
# Format results
formatted_results = [
{
"id": result.id,
"score": result.score,
"text": result.metadata.get("text", ""),
"file_path": result.metadata.get("file_path", ""),
"line_range": result.metadata.get("line_range", ""),
"type": result.metadata.get("type", "code"),
"language": result.metadata.get("language", ""),
"timestamp": result.metadata.get("timestamp", "")
}
for result in results
if result.score >= threshold
]
return {"results": formatted_results}
self.mcp_server.add_tool(
name="vector-search",
fn=vector_search,
description="Search for code snippets semantically similar to the query text"
)
logger.debug("Vector search tool registered")
def _register_knowledge(self):
"""Register the knowledge base tool with the MCP server."""
knowledge_base = self.state.get_component("knowledge_base")
if not knowledge_base:
logger.warning("Knowledge base component not available, skipping tool registration")
return
async def search_knowledge(query: str, pattern_type: str = "code", limit: int = 5):
"""Search for patterns in the knowledge base."""
logger.debug(f"MCP knowledge search request: {query=}, {pattern_type=}, {limit=}")
results = await knowledge_base.search_patterns(
query=query,
pattern_type=pattern_type,
limit=limit
)
# Format results
formatted_results = [
{
"id": result.id,
"pattern": result.pattern,
"description": result.description,
"type": result.type,
"confidence": result.confidence,
"metadata": result.metadata
}
for result in results
]
return {"results": formatted_results}
self.mcp_server.add_tool(
name="knowledge-search",
fn=search_knowledge,
description="Search for patterns in the knowledge base"
)
logger.debug("Knowledge search tool registered")
def _register_adr(self):
"""Register the ADR management tool with the MCP server."""
adr_manager = self.state.get_component("adr_manager")
if not adr_manager:
logger.warning("ADR manager component not available, skipping tool registration")
return
async def list_adrs(status: Optional[str] = None, limit: int = 10):
"""List architectural decision records."""
logger.debug(f"MCP ADR list request: {status=}, {limit=}")
try:
adrs = await adr_manager.list_adrs(status=status, limit=limit)
# Format results
formatted_results = [
{
"id": adr.id,
"title": adr.title,
"status": adr.status,
"date": adr.date.isoformat() if adr.date else None,
"authors": adr.authors,
"summary": adr.summary
}
for adr in adrs
]
return {"adrs": formatted_results}
except Exception as e:
logger.error(f"Error listing ADRs: {e}", exc_info=True)
return {"error": str(e), "adrs": []}
self.mcp_server.add_tool(
name="adr-list",
fn=list_adrs,
description="List architectural decision records"
)
logger.debug("ADR management tool registered")
def _register_task(self):
"""Register the task management tool with the MCP server."""
task_tracker = self.state.get_component("task_tracker")
if not task_tracker:
logger.warning("Task tracker component not available, skipping tool registration")
return
async def get_task_status(task_id: str):
"""Get the status of a specific task."""
logger.debug(f"MCP task status request: {task_id=}")
try:
status = await task_tracker.get_task_status(task_id)
return status
except Exception as e:
logger.error(f"Error getting task status: {e}", exc_info=True)
return {"error": str(e), "status": "unknown"}
self.mcp_server.add_tool(
name="task-status",
fn=get_task_status,
description="Get the status of a specific task"
)
logger.debug("Task management tool registered")
def get_starlette_app(self) -> Starlette:
"""Get the Starlette application for the MCP server.
Returns:
Configured Starlette application
"""
# Ensure tools are registered
self.register_tools()
# Create and return the Starlette app for SSE
if self._starlette_app is None:
self._starlette_app = create_sse_server(self.mcp_server)
return self._starlette_app
```