CLI chatbot cho document management. Scenario:
Sau bài này, bạn sẽ
- Setup project cấu trúc: mcp_client.py, mcp_server.py, main.py
- Install dependencies (uv hoặc pip)
- Verify setup chạy được
- Hiểu real-world: bạn build client OR server, không cả hai (trừ learning)
Dự án chúng ta sẽ xây
- User chat qua terminal
- Chatbot có thể đọc, edit documents stored in memory
- MCP server expose tools
read_doc_contentsvàedit_document - MCP client (trong main app) talk to server qua stdio
User (terminal)
↓
main.py (chatbot app)
├─ Anthropic client (Claude)
└─ MCP client (talk to server)
↓
mcp_server.py (FastMCP)
├─ Tool: read_doc_contents
└─ Tool: edit_document
Real-world note
- Lần đầu học
- build cả client và server để hiểu cả 2 side.
- Production
- bạn thường chọn 1 side:
- Build server: if you're service provider exposing to others
- Build client: if you're app dev consuming existing servers
Building both cùng lúc rare trong production.
Project structure
cli_project/ ├── .env # ANTHROPIC_API_KEY ├── pyproject.toml # dependencies ├── README.md # setup instructions ├── main.py # chatbot entry point ├── mcp_client.py # MCP client wrapper └── mcp_server.py # MCP server with tools
Dependencies
[project]
name = "cli-mcp-chatbot"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
"anthropic>=0.40.0",
"mcp>=1.0.0",
"python-dotenv>=1.0.0",
"pydantic>=2.0.0",
]
Install
Option 1: uv (recommended)
curl -LsSf https://astral.sh/uv/install.sh | sh
uv sync
uv run main.py
Option 2: pip
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install -e .
python main.py
uv faster, handles lockfile. Recommend.
.env
ANTHROPIC_API_KEY="sk-ant-api03-..."
Không commit. In .gitignore.
main.py skeleton
import asyncio
from dotenv import load_dotenv
from anthropic import Anthropic
from mcp_client import MCPClient
load_dotenv()
anthropic = Anthropic()
MODEL = "claude-sonnet-4-6-20260205"
async def chat_loop():
async with MCPClient(command="uv", args=["run", "mcp_server.py"]) as mcp:
print("Chatbot ready. Type 'quit' to exit.\n")
messages = []
while True:
user_input = input("You: ").strip()
if user_input.lower() in ("quit", "exit"):
break
if not user_input:
continue
messages.append({"role": "user", "content": user_input})
# Get tools from MCP
tools = await mcp.list_tools()
anthropic_tools = [tool_to_schema(t) for t in tools]
# Call Claude with tools
while True:
response = anthropic.messages.create(
model=MODEL,
max_tokens=1000,
messages=messages,
tools=anthropic_tools
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason == "end_turn":
text = "\n".join(b.text for b in response.content if b.type == "text")
print(f"Claude: {text}\n")
break
if response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f"→ Calling MCP tool: {block.name}")
result = await mcp.call_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": format_mcp_result(result)
})
messages.append({"role": "user", "content": tool_results})
def tool_to_schema(mcp_tool):
return {
"name": mcp_tool.name,
"description": mcp_tool.description,
"input_schema": mcp_tool.inputSchema
}
def format_mcp_result(result):
return "\n".join(c.text for c in result.content if c.type == "text")
if __name__ == "__main__":
asyncio.run(chat_loop())
mcp_server.py skeleton
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("DocumentMCP", log_level="ERROR")
docs = {
"report.pdf": "This report describes Q3 performance...",
"plan.md": "Project roadmap for 2026..."
}
@mcp.tool(name="hello", description="Simple hello tool for testing")
def hello(name: str) -> str:
return f"Hello, {name}!"
if __name__ == "__main__":
mcp.run(transport="stdio")
Placeholder. Full implementation bài 6.63.
mcp_client.py skeleton
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
class MCPClient:
def __init__(self, command: str, args: list[str]):
self.command = command
self.args = args
self._session = None
self._exit_stack = AsyncExitStack()
async def __aenter__(self):
params = StdioServerParameters(command=self.command, args=self.args)
read, write = await self._exit_stack.enter_async_context(stdio_client(params))
self._session = await self._exit_stack.enter_async_context(
ClientSession(read, write)
)
await self._session.initialize()
return self
async def __aexit__(self, *args):
await self._exit_stack.aclose()
async def list_tools(self):
return (await self._session.list_tools()).tools
async def call_tool(self, name: str, args: dict):
return await self._session.call_tool(name, args)
Full implementation bài 6.61.
Verify setup
uv run main.py
#
#
Nếu chạy thành công → ready for next steps.
Common setup issues
ModuleNotFoundError: No module named 'mcp'
Fix: pip install mcp hoặc uv sync.
Anthropic auth fail
Fix: Check .env loaded, key correct.
MCP server doesn't start
Fix: Check path trong args=["run", "mcp_server.py"] đúng. Server cần executable.
Áp dụng ngay
Bài tập: Clone + run (15 phút)
- Create folder
cli_project/ - Create
pyproject.toml,.env, 3 Python files - Copy skeletons
uv sync(or pip install)uv run main.py- Chat: ask "what is 2+2?"
Verify works end-to-end before proceeding.
Tóm tắt bài học
Những điều cốt lõi
- Project structure: main, mcp_client, mcp_server, .env.
- uv recommended — fast, modern Python tooling.
- Real production: build EITHER client OR server.
- Verify setup runs trước khi add real logic.
- Skeletons ở bài này — full implementation qua next lessons.
Kiểm tra nhanh1 câu hỏi · chọn đáp án để xem giải thích
Bạn đã hoàn thành bài này chưa?Tiến độ lưu trên trình duyệt của bạn.