GenAI + MCP 학습 트랙이 실제로 동작하는 RAG 및 도구 어시스턴트로 발전했습니다

# Python RAG + MCP Chat App

A minimal, readable chat app that combines retrieval-augmented generation (RAG) with MCP tool calling. Built as a companion to `microsoft/generative-ai-for-beginners` and `microsoft/mcp-for-beginners`.

## What it does

1. **Retrieval** — loads markdown from `./docs`, chunks and embeds them, retrieves top-k context per question.
2. **Chat** — calls an OpenAI-compatible chat endpoint configured via env vars.
3. **Tool loop** — supports a tool-calling loop (model → tool → model).
4. **MCP** — connects to two local MCP servers over stdio: one exposing `search_docs(query)`, one exposing a simple utility tool.
5. **Logging** — every tool call is logged with args and result.

## Lesson mapping

| Piece | Maps to |
|---|---|
| Chunking + embeddings + top-k retrieval | `generative-ai-for-beginners` — Lesson 08 (Building Search Applications) and Lesson 09 (Building RAG Applications) |
| OpenAI-compatible chat call via env vars | `generative-ai-for-beginners` — Lesson 02 (Exploring and comparing different LLMs) and Lesson 06 (Text generation apps) |
| Tool-calling loop | `generative-ai-for-beginners` — Lesson 11 (Integrating with function calling) |
| MCP client over stdio | `mcp-for-beginners` — Lesson 03 (Getting started with MCP) and Lesson 04 (Building an MCP client) |
| MCP server exposing a tool | `mcp-for-beginners` — Lesson 02 (MCP core concepts) and Lesson 05 (Building an MCP server) |
| Tool-call logging | `mcp-for-beginners` — Lesson 06 (Observability and logging) |

## Project layout

```
.
├── README.md
├── requirements.txt
├── .env.example
├── app.py
├── rag.py
├── mcp_client.py
├── servers/
│   ├── docs_server.py
│   └── util_server.py
└── docs/
    └── (your markdown files)
```

## Setup

```bash
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env             # then edit .env
```

Drop your markdown files into `./docs`. Any `.md` file is picked up.

## Environment variables

| Var | Purpose |
|---|---|
| `OPENAI_BASE_URL` | Base URL of an OpenAI-compatible endpoint (e.g. `https://api.openai.com/v1`, or a local server) |
| `OPENAI_API_KEY` | API key for that endpoint |
| `CHAT_MODEL` | Chat model name (e.g. `gpt-4o-mini`) |
| `EMBED_MODEL` | Embedding model name (e.g. `text-embedding-3-small`) |
| `TOP_K` | Number of chunks to retrieve (default `4`) |
| `DOCS_DIR` | Docs directory (default `./docs`) |

## Run

```bash
python app.py
```

The app starts both MCP servers as subprocesses over stdio, builds the retrieval index, then enters a chat loop.

## Example prompts

These exercise both retrieval and tools:

1. **Retrieval only** — `Summarize the key ideas in the docs about RAG.`
   Triggers `search_docs` via the MCP docs server, then answers from the retrieved chunks.

2. **Retrieval + utility tool** — `How many words are in the docs about MCP? Use the utility tool to count.`
   Exercises `search_docs` and the utility tool in the same turn, showing the tool-calling loop.

3. **Multi-step** — `Find the section on function calling, then compute the character count of the retrieved text.`
   Chains retrieval and the utility tool, and logs both calls with args and results.

## Files

### `requirements.txt`

```
openai>=1.40.0
mcp>=1.0.0
numpy>=1.26.0
python-dotenv>=1.0.0
```

### `.env.example`

```
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_API_KEY=sk-...
CHAT_MODEL=gpt-4o-mini
EMBED_MODEL=text-embedding-3-small
TOP_K=4
DOCS_DIR=./docs
```

### `rag.py`

```python
"""Chunk, embed, and retrieve markdown docs.

Maps to generative-ai-for-beginners Lesson 08 (search) and Lesson 09 (RAG).
"""
import os
from pathlib import Path
from typing import List, Tuple

import numpy as np
from openai import OpenAI

CHUNK_SIZE = 800
CHUNK_OVERLAP = 100


def _client() -> OpenAI:
    return OpenAI(
        base_url=os.environ["OPENAI_BASE_URL"],
        api_key=os.environ["OPENAI_API_KEY"],
    )


def _chunk(text: str, size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> List[str]:
    chunks, start = [], 0
    while start < len(text):
        end = min(start + size, len(text))
        chunks.append(text[start:end])
        if end == len(text):
            break
        start = end - overlap
    return chunks


def _embed(texts: List[str]) -> np.ndarray:
    resp = _client().embeddings.create(model=os.environ["EMBED_MODEL"], input=texts)
    return np.array([d.embedding for d in resp.data], dtype=np.float32)


class Retriever:
    def __init__(self, docs_dir: str):
        self.chunks: List[Tuple[str, str]] = []  # (source, text)
        for path in Path(docs_dir).rglob("*.md"):
            for c in _chunk(path.read_text(encoding="utf-8")):
                self.chunks.append((str(path), c))
        if not self.chunks:
            raise RuntimeError(f"No markdown files found in {docs_dir}")
        self.vectors = _embed([c for _, c in self.chunks])
        # normalize for cosine similarity
        self.vectors /= np.linalg.norm(self.vectors, axis=1, keepdims=True) + 1e-8

    def search(self, query: str, k: int = 4) -> List[Tuple[str, str, float]]:
        q = _embed([query])[0]
        q /= np.linalg.norm(q) + 1e-8
        scores = self.vectors @ q
        idx = np.argsort(-scores)[:k]
        return [(self.chunks[i][0], self.chunks[i][1], float(scores[i])) for i in idx]
```

### `mcp_client.py`

```python
"""Minimal MCP client over stdio.

Maps to mcp-for-beginners Lesson 03 (getting started) and Lesson 04 (client).
"""
import json
import sys
from contextlib import AsyncExitStack
from typing import Any, Dict, List

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


class MCPConnection:
    def __init__(self, name: str, command: str, args: List[str]):
        self.name = name
        self.params = StdioServerParameters(command=command, args=args)
        self._stack = AsyncExitStack()
        self.session: ClientSession | None = None
        self.tools: List[Dict[str, Any]] = []

    async def __aenter__(self):
        read, write = await self._stack.enter_async_context(stdio_client(self.params))
        self.session = await self._stack.enter_async_context(ClientSession(read, write))
        await self.session.initialize()
        listed = await self.session.list_tools()
        self.tools = [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description or "",
                    "parameters": t.inputSchema or {"type": "object", "properties": {}},
                },
            }
            for t in listed.tools
        ]
        return self

    async def __aexit__(self, *exc):
        await self._stack.aclose()

    async def call(self, tool: str, args: Dict[str, Any]) -> str:
        result = await self.session.call_tool(tool, args)
        # MCP returns a list of content blocks; flatten text
        parts = []
        for block in result.content:
            text = getattr(block, "text", None)
            if text is not None:
                parts.append(text)
            else:
                parts.append(json.dumps(block.model_dump(), default=str))
        return "\n".join(parts)
```

### `servers/docs_server.py`

```python
"""MCP server exposing search_docs(query).

Maps to mcp-for-beginners Lesson 02 (core concepts) and Lesson 05 (server).
"""
import os
from pathlib import Path

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("docs-server")
DOCS_DIR = os.environ.get("DOCS_DIR", "./docs")


@mcp.tool()
def search_docs(query: str, k: int = 4) -> str:
    """Search markdown docs in ./docs for the given query and return top-k chunks."""
    # Simple keyword scoring keeps the server dependency-free.
    terms = [t.lower() for t in query.split() if t.strip()]
    scored = []
    for path in Path(DOCS_DIR).rglob("*.md"):
        text = path.read_text(encoding="utf-8")
        score = sum(text.lower().count(t) for t in terms)
        if score:
            scored.append((score, str(path), text[:1200]))
    scored.sort(reverse=True)
    if not scored:
        return "No matches."
    return "\n\n---\n\n".join(f"[{p}]\n{t}" for _, p, t in scored[:k])


if __name__ == "__main__":
    mcp.run()
```

### `servers/util_server.py`

```python
"""MCP server exposing a simple utility tool.

Maps to mcp-for-beginners Lesson 05 (server) and Lesson 06 (observability).
"""
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("util-server")


@mcp.tool()
def count_text(text: str) -> str:
    """Return character and word counts for the given text."""
    return f"chars={len(text)} words={len(text.split())}"


if __name__ == "__main__":
    mcp.run()
```

### `app.py`

```python
"""Chat app: RAG + tool-calling loop + MCP over stdio.

Maps to:
- generative-ai-for-beginners Lesson 06 (chat apps), Lesson 09 (RAG), Lesson 11 (function calling)
- mcp-for-beginners Lesson 04 (client), Lesson 06 (logging)
"""
import asyncio
import json
import os
import sys

from dotenv import load_dotenv
from openai import OpenAI

from mcp_client import MCPConnection
from rag import Retriever

load_dotenv()

SYSTEM_PROMPT = (
    "You are a helpful assistant. Use the provided tools when they help. "
    "Prefer search_docs for questions about the local documentation."
)


def log_tool_call(server: str, tool: str, args: dict, result: str) -> None:
    # Lesson 06 (mcp-for-beginners): log every tool call with args and result.
    print(f"[tool] {server}.{tool} args={json.dumps(args)} -> {result[:200]!r}", file=sys.stderr)


async def main() -> None:
    retriever = Retriever(os.environ.get("DOCS_DIR", "./docs"))
    top_k = int(os.environ.get("TOP_K", "4"))

    async with MCPConnection(
        "docs", sys.executable, ["servers/docs_server.py"]
    ) as docs_conn, MCPConnection(
        "util", sys.executable, ["servers/util_server.py"]
    ) as util_conn:
        # Map tool name -> (server name, connection)
        tool_index = {}
        all_tools = []
        for conn in (docs_conn, util_conn):
            for t in conn.tools:
                name = t["function"]["name"]
                tool_index[name] = (conn.name, conn)
                all_tools.append(t)

        client = OpenAI(
            base_url=os.environ["OPENAI_BASE_URL"],
            api_key=os.environ["OPENAI_API_KEY"],
        )
        model = os.environ["CHAT_MODEL"]

        messages = [{"role": "system", "content": SYSTEM_PROMPT}]
        print("Chat ready. Type 'exit' to quit.")

        while True:
            try:
                user = input("\nYou: ").strip()
            except (EOFError, KeyboardInterrupt):
                break
            if not user or user.lower() == "exit":
                break

            # Retrieval-augmented context (Lesson 09).
            hits = retriever.search(user, k=top_k)
            context = "\n\n".join(f"[{src}]\n{chunk}" for src, chunk, _ in hits)
            messages.append(
                {
                    "role": "user",
                    "content": f"Context:\n{context}\n\nQuestion: {user}",
                }
            )

            # Tool-calling loop (Lesson 11).
            while True:
                resp = client.chat.completions.create(
                    model=model,
                    messages=messages,
                    tools=all_tools,
                    tool_choice="auto",
                )
                msg = resp.choices[0].message
                messages.append(msg.model_dump(exclude_none=True))

                if not msg.tool_calls:
                    print(f"\nAssistant: {msg.content}")
                    break

                for call in msg.tool_calls:
                    name = call.function.name
                    args = json.loads(call.function.arguments or "{}")
                    server_name, conn = tool_index[name]
                    result = await conn.call(name, args)
                    log_tool_call(server_name, name, args, result)
                    messages.append(
                        {
                            "role": "tool",
                            "tool_call_id": call.id,
                            "content": result,
                        }
                    )


if __name__ == "__main__":
    asyncio.run(main())
```

## Notes

- The docs MCP server uses keyword scoring to stay dependency-free; swap in the `Retriever` from `rag.py` if you want embeddings inside the server too.
- Both servers are launched with `sys.executable` so they use the same virtualenv.
- Tool calls are logged to stderr so they don't pollute the chat transcript on stdout.
- To add a third MCP server, drop a file in `servers/`, add an `MCPConnection(...)` in `app.py`, and it will be picked up automatically.