ドキュメント検索MCPサーバを作ってみた【MCP+OpenSearch+AWS】

backend
├── Dockerfile
├── pyproject.toml
├── src
│   └── documentation_search
│       ├── __init__.py
│       ├── application
│       │   ├── __init__.py
│       │   ├── dtos
│       │   │   ├── __init__.py
│       │   │   └── search.py
│       │   ├── exceptions.py
│       │   └── usecases
│       │       ├── __init__.py
│       │       └── search_documents.py
│       ├── domain
│       │   ├── __init__.py
│       │   ├── aggregates
│       │   │   ├── __init__.py
│       │   │   └── search.py
│       │   ├── entities
│       │   │   ├── __init__.py
│       │   │   ├── query_result.py
│       │   │   └── query.py
│       │   ├── exceptions.py
│       │   ├── repositories
│       │   │   ├── __init__.py
│       │   │   └── search.py
│       │   └── value_objects
│       │       ├── __init__.py
│       │       ├── document_suggestion.py
│       │       ├── query_status.py
│       │       └── search_criteria_value_object.py
│       ├── infrastructure
│       │   ├── __init__.py
│       │   ├── config
│       │   │   ├── __init__.py
│       │   │   └── opensearch_config.py
│       │   └── persistence
│       │       ├── __init__.py
│       │       ├── in_memory_search.py
│       │       └── opensearch_search.py
│       ├── main.py
│       └── presentation
│           ├── __init__.py
│           ├── dependencies.py
│           └── routes
│               ├── __init__.py
│               └── search.py
├── tests
└── uv.lock

"""Search criteria"""

from humps import camelize
from pydantic import BaseModel


class TextQuery(BaseModel):
    """テキストクエリ"""

    class Config:
        alias_generator = camelize
        populate_by_name = True

    input: str


class SearchCriteria(BaseModel):
    """検索条件"""

    class Config:
        alias_generator = camelize
        populate_by_name = True

    text_query: TextQuery

"""Document suggestion"""

from humps import camelize
from pydantic import BaseModel, Field


class TextExcerptSuggestion(BaseModel):
    """検索結果のテキスト抜粋"""

    class Config:
        alias_generator = camelize
        populate_by_name = True

    link: str = Field(description='ドキュメントへのリンクURL')
    title: str = Field(description='ドキュメントのタイトル')
    suggestion_body: str = Field(description='提案内容の本文')
    summary: str = Field(description='ドキュメントの要約')
    source_updated_at: int = Field(description='ソースの更新日時 (UNIX timestamp in milliseconds)')


class Suggestion(BaseModel):
    """検索結果の提案"""

    class Config:
        alias_generator = camelize
        populate_by_name = True

    text_excerpt_suggestion: TextExcerptSuggestion = Field(description='テキスト抜粋提案')

"""OpenSearch search repository implementation."""

from documentation_search.domain.repositories.search import ISearchRepository
from documentation_search.domain.value_objects.document_suggestion import (
    Suggestion,
    TextExcerptSuggestion,
)
from documentation_search.domain.value_objects.search_criteria_value_object import (
    SearchCriteria,
)
from opensearchpy import OpenSearch
from typing import List


class OpenSearchSearchRepository(ISearchRepository):
    """OpenSearch検索リポジトリの実装."""

    def __init__(
        self,
        client: OpenSearch,
        index_name: str,
    ) -> None:
        self._client = client
        self._index_name = index_name

    def find_matching_documents(self, criteria: SearchCriteria) -> List[Suggestion]:
        """検索条件に一致するドキュメントを検索する.

        OpenSearchを使用して全文検索を実行する。
        """
        query = {
            'query': {
                'multi_match': {
                    'query': criteria.text_query.input,
                    'fields': ['title^2', 'body', 'summary'],
                    'type': 'best_fields',
                    'fuzziness': 'AUTO',
                }
            },
            'size': 10,
            'highlight': {
                'fields': {
                    'body': {
                        'fragment_size': 200,
                        'number_of_fragments': 3,
                        'pre_tags': [''],
                        'post_tags': [''],
                    }
                },
                'order': 'score',
            },
        }

        try:
            response = self._client.search(index=self._index_name, body=query)

            suggestions = []
            for hit in response['hits']['hits']:
                source = hit['_source']

                suggestion_body = source.get('body', '')
                if 'highlight' in hit and 'body' in hit['highlight']:
                    # ハイライトフラグメントを結合（最大3つのフラグメント）
                    fragments = hit['highlight']['body']
                    suggestion_body = '... '.join(fragments)
                else:
                    # ハイライトがない場合は先頭200文字を使用
                    suggestion_body = (
                        suggestion_body[:200] + '...'
                        if len(suggestion_body) > 200
                        else suggestion_body
                    )

                text_excerpt_suggestion = TextExcerptSuggestion(
                    link=source.get('link', ''),
                    title=source.get('title', ''),
                    suggestion_body=suggestion_body,
                    summary=source.get('summary', ''),
                    source_updated_at=source.get('source_updated_at', 0),
                )
                suggestions.append(Suggestion(text_excerpt_suggestion=text_excerpt_suggestion))

            return suggestions

        except Exception as e:
            # エラーログを出力して空のリストを返す
            print(f'Error searching OpenSearch: {e}')
            return []

import httpx
from mcp.server.fastmcp import FastMCP, Context
from pydantic import Field
from typing import List


SEARCH_API_URL = 'https://tidb.search.docs.example.com/search'

mcp = FastMCP(
    'lv-levtech.tidb-documentation-mcp-server',
    instructions="""
    TiDB Documentation MCP Server

    このサーバーは TiDB の公式ドキュメントを検索します。
    """
)


@mcp.tool()
async def search_documentation(
    ctx: Context,
    search_phrase: str = Field(description='検索フレーズ（例：分散トランザクション）'),
    limit: int = Field(default=5, description='検索結果の最大件数', ge=1, le=20)
) -> str:
    """TiDB ドキュメントを検索する

    Args:
        query: 検索キーワード
        limit: 結果件数

    Returns:
        検索結果
    """
    request_body = {
        'textQuery': {
            'input': search_phrase,
        },
    }
    async with httpx.AsyncClient() as client:
        response = await client.post(
            SEARCH_API_URL,
            json=request_body,
            timeout=30.0
        )
        response.raise_for_status()
        data = response.json()

    if not data.get('suggestions'):
        return '検索結果が見つかりませんでした。'

    results = []
    for i, suggestion in enumerate(data['suggestions'][:limit]):
        if 'textExcerptSuggestion' in suggestion:
            text_suggestion = suggestion['textExcerptSuggestion']
            context = None

            if 'summary' in text_suggestion:
                context = text_suggestion['summary']
            elif 'suggestionBody' in text_suggestion:
                context = text_suggestion['suggestionBody']

            results.append(
                SearchResult(
                    rank_order=i + 1,
                    url=text_suggestion.get('link', ''),
                    title=text_suggestion.get('title', ''),
                    query_id=data.get('queryId', ''),
                    context=context,
                )
            )

    return results


if __name__ == '__main__':
    mcp.run()

class Indexer:
    """ドキュメントリポジトリをOpenSearchにインデックスするクラス"""

    def index_repository(self, repo_url: str, base_url: str):
        # 1. GitHubリポジトリをクローン
        repo = self.clone_repository(repo_url)

        # 2. Markdownファイルを検索
        markdown_files = self.find_markdown_files(repo_root)

        # 3. 各ファイルをパース
        documents = []
        for file_path in markdown_files:
            doc = self.prepare_document(file_path)
            documents.append(doc)

        # 4. OpenSearchにバルクインデックス
        self.bulk_index(documents)


def parse_markdown_file(file_path: Path) -> Dict:
    """Markdownファイルをパースする"""
    with open(file_path, 'r') as f:
        post = frontmatter.load(f)  # frontmatterを抽出

    # メタデータを取得
    title = post.metadata.get('title', '')
    summary = post.metadata.get('summary', '')

    # Markdown → HTML → プレーンテキスト変換
    html = markdown.markdown(post.content)
    plain_text = BeautifulSoup(html).get_text()

    return {
        'title': title,
        'summary': summary,
        'body': post.content,
        'link': generate_url(file_path),  # ファイルパスからURL生成
        'source_updated_at': int(time.time() * 1000),
    }