# Code Search MCP 完整使用示例

本文档提供 Code Search MCP 的完整使用示例，从基础到高级。

---

## 1. 基础索引示例

### 1.1 首次全量索引

```python
from mcp import ClientSession, StdioServerParameters
import asyncio

async def first_index():
    params = StdioServerParameters(
        command="python",
        args=["-m", "code_search_mcp"]
    )

    async with ClientSession(params) as session:
        await session.initialize()

        # 首次索引 - 全量
        result = await session.call_tool("index_repository", {
            "repo_path": "/Users/zeng/my-project",
            "force_rebuild": True  # 首次全量索引
        })

        print(f"索引完成: {result}")
        # {
        #   "success": True,
        #   "stats": {
        #     "files_indexed": 150,
        #     "files_skipped": 0,
        #     "files_deleted": 0,
        #     "symbols": 1200,
        #     "content_blocks": 5000,
        #     "elapsed_seconds": 0.8
        #   }
        # }

asyncio.run(first_index())
```

### 1.2 增量索引

```python
async def incremental_index():
    async with ClientSession(params) as session:
        await session.initialize()

        # 后续更新 - 增量
        result = await session.call_tool("index_repository", {
            "repo_path": "/Users/zeng/my-project",
            "force_rebuild": False  # 增量，仅更新变更文件
        })

        print(f"增量索引: {result['stats']['files_indexed']} 文件更新")
        # 100 文件通常 < 0.1 秒
```

---

## 2. 搜索示例

### 2.1 文件搜索

```python
async def search_files():
    async with ClientSession(params) as session:
        await session.initialize()

        # 搜索配置文件
        result = await session.call_tool("find_files", {
            "query": "config",
            "extensions": [".json", ".yaml", ".yml", ".toml"],
            "limit": 20
        })

        for f in result.get("files", []):
            print(f"  {f['path']}")
```

### 2.2 符号搜索

```python
async def search_symbols():
    async with ClientSession(params) as session:
        await session.initialize()

        # 搜索函数
        result = await session.call_tool("find_symbols", {
            "query": "handle",
            "kind": "function",
            "limit": 30
        })

        for s in result.get("symbols", []):
            print(f"  {s['name']} @ {s['file']}:{s['line']}")

        # 搜索类
        result = await session.call_tool("find_symbols", {
            "query": "Service",
            "kind": "class",
            "limit": 20
        })
```

### 2.3 内容搜索

```python
async def search_content():
    async with ClientSession(params) as session:
        await session.initialize()

        # 普通搜索
        result = await session.call_tool("find_content", {
            "query": "TODO",
            "limit": 50
        })

        # 正则搜索 - 找所有函数定义
        result = await session.call_tool("find_content", {
            "query": "^def\\s+\\w+\\s*\\(',
            "regex": True,
            "file_pattern": "*.py",
            "limit": 30
        })

        # 找 async 函数
        result = await session.call_tool("find_content", {
            "query": "async def",
            "file_pattern": "*.ts",
            "limit": 20
        })
```

---

## 3. 文件结构分析

```python
async def analyze_file():
    async with ClientSession(params) as session:
        await session.initialize()

        # 获取文件结构
        result = await session.call_tool("get_structure", {
            "file_path": "/Users/zeng/project/src/main.py"
        })

        print(f"文件: {result['path']}")
        print(f"语言: {result['language']}")
        print("符号:")
        for s in result.get("symbols", []):
            print(f"  {s['kind']}: {s['name']} @ line {s['line']}")
```

---

## 4. 统计和监控

### 4.1 获取索引统计

```python
async def get_stats():
    async with ClientSession(params) as session:
        await session.initialize()

        result = await session.call_tool("get_stats", {})

        stats = result["stats"]
        print(f"文件: {stats['files']}")
        print(f"符号: {stats['symbols']}")
        print(f"内容块: {stats['content_blocks']}")
```

### 4.2 获取调用统计

```python
async def get_call_stats():
    async with ClientSession(params) as session:
        await session.initialize()

        result = await session.call_tool("get_call_stats", {})

        stats = result["call_stats"]
        print(f"总调用: {stats['total_calls']}")
        print(f"成功率: {stats['success_rate']}%")
        print(f"平均耗时: {stats['avg_duration_ms']:.2f}ms")

        print("\n按工具统计:")
        for tool in stats["by_tool"]:
            print(f"  {tool['tool_name']}: {tool['count']} 次, "
                  f"平均 {tool['avg_duration']:.2f}ms")
```

### 4.3 查看日志

```python
async def view_logs():
    async with ClientSession(params) as session:
        await session.initialize()

        # 查看最近日志
        result = await session.call_tool("get_logs", {
            "limit": 20
        })

        # 查看错误日志
        result = await session.call_tool("get_logs", {
            "success": False,
            "limit": 10
        })

        # 查看特定工具日志
        result = await session.call_tool("get_logs", {
            "tool_name": "find_files",
            "limit": 50
        })
```

---

## 5. 完整工作流

### 5.1 代码探索工作流

```python
async def explore_codebase(repo_path: str):
    """完整代码库探索工作流"""

    params = StdioServerParameters(
        command="python",
        args=["-m", "code_search_mcp"]
    )

    async with ClientSession(params) as session:
        await session.initialize()

        # Step 1: 索引
        print("📚 索引代码库...")
        index_result = await session.call_tool("index_repository", {
            "repo_path": repo_path,
            "force_rebuild": False
        })

        if not index_result.get("success"):
            print(f"索引失败: {index_result}")
            return

        # Step 2: 获取统计
        stats_result = await session.call_tool("get_stats", {})
        stats = stats_result["stats"]
        print(f"✅ 索引完成: {stats['files']} 文件, "
              f"{stats['symbols']} 符号")

        # Step 3: 搜索文件
        print("\n🔍 搜索文件...")
        files_result = await session.call_tool("find_files", {
            "query": "index",
            "limit": 10
        })
        print(f"找到 {len(files_result['files'])} 个文件")

        # Step 4: 搜索符号
        print("\n🔍 搜索符号...")
        symbols_result = await session.call_tool("find_symbols", {
            "query": "",
            "kind": "class",
            "limit": 10
        })
        print(f"找到 {len(symbols_result['symbols'])} 个类")

        # Step 5: 搜索内容
        print("\n🔍 搜索内容...")
        content_result = await session.call_tool("find_content", {
            "query": "TODO",
            "limit": 20
        })
        print(f"找到 {len(content_result['matches'])} 处 TODO")

        # Step 6: 检查性能
        perf_result = await session.call_tool("get_call_stats", {})
        perf = perf_result["call_stats"]
        print(f"\n📊 性能: {perf['total_calls']} 次调用, "
              f"平均 {perf['avg_duration_ms']:.2f}ms")


# 运行工作流
asyncio.run(explore_codebase("/Users/zeng/my-project"))
```

### 5.2 Bug 定位工作流

```python
async def find_bug(keyword: str):
    """Bug 定位工作流"""

    async with ClientSession(params) as session:
        await session.initialize()

        # Step 1: 搜索错误相关代码
        print(f"🔍 搜索 '{keyword}'...")
        result = await session.call_tool("find_content", {
            "query": keyword,
            "limit": 50
        })

        matches = result.get("matches", [])
        print(f"找到 {len(matches)} 处匹配")

        # Step 2: 分析匹配位置
        file_occurrences = {}
        for m in matches:
            f = m["file"]
            if f not in file_occurrences:
                file_occurrences[f] = []
            file_occurrences[f].append(m["line"])

        # Step 3: 获取文件结构
        for file_path, lines in file_occurrences.items()[:3]:
            print(f"\n📄 {file_path} (行 {lines[:5]})")
            struct = await session.call_tool("get_structure", {
                "file_path": file_path
            })
            for s in struct.get("symbols", [])[:5]:
                print(f"  - {s['kind']}: {s['name']}")
```

---

## 6. 并行搜索

```python
async def parallel_search():
    """并行执行多个搜索"""

    async with ClientSession(params) as session:
        await session.initialize()

        # 并行搜索
        files, symbols, content, stats = await asyncio.gather(
            session.call_tool("find_files", {"query": "test", "limit": 10}),
            session.call_tool("find_symbols", {"query": "class", "limit": 10}),
            session.call_tool("find_content", {"query": "TODO", "limit": 10}),
            session.call_tool("get_stats", {})
        )

        print(f"文件: {len(files['files'])}")
        print(f"符号: {len(symbols['symbols'])}")
        print(f"内容: {len(content['matches'])}")
        print(f"索引: {stats['stats']['files']} 文件")
```

---

## 7. 定期增量更新

```python
import schedule
import time

async def daily_update(repo_path: str):
    """每日增量更新"""

    async with ClientSession(params) as session:
        await session.initialize()

        result = await session.call_tool("index_repository", {
            "repo_path": repo_path,
            "force_rebuild": False  # 增量
        })

        stats = result.get("stats", {})
        print(f"增量更新: {stats.get('files_indexed', 0)} 新增, "
              f"{stats.get('files_skipped', 0)} 跳过")

# 定时任务
schedule.every().day.at("09:00").do(
    lambda: asyncio.run(daily_update("/path/to/repo"))
)

while True:
    schedule.run_pending()
    time.sleep(60)
```

---

## 8. CLI 使用示例

```bash
# 索引
code-search index ~/my-project
code-search rebuild ~/my-project  # 全量

# 搜索
code-search search "TODO"
code-search files "config"
code-search symbols "handle"
code-search content "async def"

# 监控
code-search logs
code-search logs --errors-only
code-search stats
code-search status

# 清理
code-search clean --logs 7    # 清理7天前日志
code-search clean --index    # 清理索引
code-search clean --all      # 清理全部
```
