도구

!uv pip install fastapi "uvicorn[standard]" httpx fastmcp nest-asyncio

import sqlite3

def init_db():
    conn = sqlite3.connect("books.db")
    cursor = conn.cursor()

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS books (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        author TEXT NOT NULL,
        category TEXT NOT NULL,
        summary TEXT
    )
    """)

    sample_books = [
        ("딥러닝 언어 모델 이론", "이성주", "인공지능", "트랜스포머 아키텍처부터 사전훈련 모델 활용까지를 다룹니다."),
        ("Ollama로 시작하는 로컬 LLM", "김철수", "인공지능", "초보자를 위한 Ollama 환경 설정과 API 연동 가이드."),
        ("파이썬 알고리즘 프로그래밍", "박영희", "컴퓨터공학", "자료구조와 코딩테스트 기출문제 풀이.")
    ]

    cursor.execute("SELECT COUNT(*) FROM books")
    if cursor.fetchone()[0] == 0:
        cursor.executemany("""
        INSERT INTO books (title, author, category, summary)
        VALUES (?, ?, ?, ?)
        """, sample_books)
        conn.commit()
        print("books.db에 샘플 데이터 입력이 성공적으로 수행되었습니다.")
    else:
        print("books.db 데이터베이스가 이미 존재하며 데이터가 채워져 있습니다.")

    conn.close()

init_db()

%%writefile book_api.py
import sqlite3
from fastapi import FastAPI

app = FastAPI(title="Book API")

def query_books(query: str) -> str:
    conn = sqlite3.connect("books.db")
    cursor = conn.cursor()
    cursor.execute(
        "SELECT title, author, category, summary FROM books WHERE title LIKE ? OR category LIKE ?",
        (f"%{query}%", f"%{query}%"),
    )
    rows = cursor.fetchall()
    conn.close()

    if not rows:
        return f"'{query}'에 대한 도서 검색 결과가 없습니다."

    result_str = "[도서 검색 결과]\n"
    for title, author, category, summary in rows:
        result_str += f"- 제목: {title} | 저자: {author} | 분류: {category}\n  요약: {summary}\n"
    return result_str.strip()

@app.get(
    "/search",
    operation_id="search_books",
    description="도서 데이터베이스에서 제목 또는 카테고리에 검색어가 포함된 책을 찾습니다. "
                "query에는 사용자 질문의 한국어 키워드를 그대로 넣습니다(영어로 번역하지 않습니다).",
)
def search(query: str):
    return {"result": query_books(query)}

import sys
import time
import subprocess
import httpx

server = subprocess.Popen(
    [sys.executable, "-m", "uvicorn", "book_api:app", "--port", "8000"],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL,
)

# 서버가 응답할 때까지 최대 15초 대기
for _ in range(30):
    try:
        httpx.get("http://localhost:8000/openapi.json", timeout=1)
        print("FastAPI 서버가 준비되었습니다: http://localhost:8000")
        break
    except Exception:
        time.sleep(0.5)
else:
    print("서버 시작에 실패했습니다. 포트 충돌 여부를 확인하세요.")

import os
import httpx

# Ollama 서버 주소. localhost가 아니면(예: WSL에서 Windows 호스트) OLLAMA_HOST로 지정합니다.
OLLAMA_API_URL = f"{os.getenv('OLLAMA_HOST', 'http://localhost:11434')}/api/chat"
MODEL_NAME = "qwen3:latest"  # 도구 호출(tool calling)을 지원하는 모델이 필요합니다. gemma 계열은 미지원이므로 qwen3를 사용합니다.
BOOK_API_URL = "http://localhost:8000"

# 도구 스키마를 손으로 정의합니다 (Ollama function-calling 형식).
tools = [{
    "type": "function",
    "function": {
        "name": "search_books",
        "description": "도서 데이터베이스에서 제목 또는 카테고리에 쿼리가 포함된 도서를 검색합니다.",
        "parameters": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "검색어"}},
            "required": ["query"],
        },
    },
}]

# 도구 이름을 보고 해당 REST 엔드포인트로 직접 연결합니다 (수작업 디스패치).
def call_tool(name, arguments):
    if name == "search_books":
        response = httpx.get(f"{BOOK_API_URL}/search", params={"query": arguments["query"]}, timeout=30)
        return response.json()["result"]
    return f"알 수 없는 도구입니다: {name}"

def chat(messages, with_tools=False):
    payload = {"model": MODEL_NAME, "messages": messages, "stream": False}
    if with_tools:
        payload["tools"] = tools
    response = httpx.post(OLLAMA_API_URL, json=payload, timeout=120)
    return response.json()["message"]

messages = [{"role": "user", "content": "인공지능 관련 도서를 데이터베이스에서 추천해줘."}]

message_obj = chat(messages, with_tools=True)
if message_obj.get("tool_calls"):
    tool_call = message_obj["tool_calls"][0]["function"]
    print(f"[도구 호출] {tool_call['name']}({tool_call['arguments']})")

    result_text = call_tool(tool_call["name"], tool_call["arguments"])
    messages.append(message_obj)
    messages.append({"role": "tool", "name": tool_call["name"], "content": result_text})

    final_message = chat(messages)
    print(f"\n[최종 답변]\n{final_message['content']}")
else:
    print(f"[도구 미사용]\n{message_obj['content']}")

%%writefile book_api.py
import os
import sqlite3
from fastapi import FastAPI
from fastmcp import FastMCP   # 'pip install fastmcp' 패키지 (아래 note 참고)

app = FastAPI(title="Book API")

def query_books(query: str) -> str:
    conn = sqlite3.connect("books.db")
    cursor = conn.cursor()
    cursor.execute(
        "SELECT title, author, category, summary FROM books WHERE title LIKE ? OR category LIKE ?",
        (f"%{query}%", f"%{query}%"),
    )
    rows = cursor.fetchall()
    conn.close()

    if not rows:
        return f"'{query}'에 대한 도서 검색 결과가 없습니다."

    result_str = "[도서 검색 결과]\n"
    for title, author, category, summary in rows:
        result_str += f"- 제목: {title} | 저자: {author} | 분류: {category}\n  요약: {summary}\n"
    return result_str.strip()

@app.get(
    "/search",
    operation_id="search_books",
    description="도서 데이터베이스에서 제목 또는 카테고리에 검색어가 포함된 책을 찾습니다. "
                "query에는 사용자 질문의 한국어 키워드를 그대로 넣습니다(영어로 번역하지 않습니다).",
)
def search(query: str):
    return {"result": query_books(query)}

# 위 FastAPI 앱은 uvicorn으로 그대로 REST 서빙할 수 있고,
# 이 파일을 직접 실행할 때만 같은 앱을 MCP 서버로 변환해 띄웁니다.
if __name__ == "__main__":
    mcp = FastMCP.from_fastapi(app=app)   # 엔드포인트 → MCP 도구 자동 변환

    # 기본은 stdio(로컬). 배포 시 MCP_TRANSPORT=http 로 주면 원격용 HTTP 서버로 실행됩니다.
    if os.getenv("MCP_TRANSPORT") == "http":
        mcp.run(transport="http", host="0.0.0.0", port=int(os.getenv("MCP_PORT", "8001")))
    else:
        mcp.run()

from fastmcp import FastMCP

mcp = FastMCP("BookStore")

@mcp.tool()
def search_books(query: str) -> str:
    """도서 데이터베이스에서 제목 또는 카테고리에 검색어가 포함된 책을 찾습니다.

    Args:
        query: 검색할 한국어 키워드 (예: '인공지능', '딥러닝')
    """
    return query_books(query)

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

import os
import asyncio
import httpx
from fastmcp import Client

# Ollama 서버 주소. 기본은 localhost이며, 예컨대 WSL에서 Windows 호스트의 Ollama를
# 쓸 때처럼 다른 곳에 있으면 OLLAMA_HOST 환경 변수로 지정합니다.
OLLAMA_HOST = os.getenv("OLLAMA_HOST", "http://localhost:11434")
OLLAMA_API_URL = f"{OLLAMA_HOST}/api/chat"
MODEL_NAME = "qwen3:latest"

# Ollama Chat 호출 유틸리티 (MCP가 알려 준 도구 목록을 Ollama 형식으로 변환)
async def call_ollama(messages, tools=None):
    payload = {"model": MODEL_NAME, "messages": messages, "stream": False}
    if tools:
        payload["tools"] = [{
            "type": "function",
            "function": {"name": t.name, "description": t.description, "parameters": t.inputSchema},
        } for t in tools]
    async with httpx.AsyncClient() as http:
        response = await http.post(OLLAMA_API_URL, json=payload, timeout=120.0)
        response.raise_for_status()
        return response.json()["message"]

async def run_agent(user_question: str):
    print(f"[User]: {user_question}\n")

    # fastmcp.Client는 'book_api.py' 경로를 보고 stdio MCP 서버로 자동 구동합니다.
    # (StdioServerParameters·ClientSession·initialize 보일러플레이트가 한 줄로 줄어듭니다.)
    async with Client("book_api.py") as client:
        tools = await client.list_tools()
        print(f"-> MCP 서버가 제공하는 도구 목록: {[t.name for t in tools]}")

        messages = [{"role": "user", "content": user_question}]

        # 모델이 더 이상 도구를 요청하지 않을 때까지 반복합니다.
        # 한 번의 응답에 여러 tool_calls가 담길 수 있으므로 전체를 순회하고,
        # range로 최대 단계를 제한해 무한 루프를 방지합니다.
        for step in range(5):
            print(f"-> Ollama 분석 요청 (step {step + 1})...")
            message_obj = await call_ollama(messages, tools)
            messages.append(message_obj)

            tool_calls = message_obj.get("tool_calls")
            if not tool_calls:
                break  # 도구 호출이 없으면 최종 답변이 나온 것

            for tool_call in tool_calls:
                tool_name = tool_call["function"]["name"]
                tool_args = tool_call["function"]["arguments"]
                print(f"\n[Agent Action] 도구 실행: {tool_name}({tool_args})")

                # Ollama는 arguments를 (OpenAI와 달리) JSON 문자열이 아닌 dict로 돌려주므로 그대로 넘깁니다.
                mcp_result = await client.call_tool(tool_name, tool_args)
                result_text = mcp_result.data["result"]   # search_books는 {"result": ...}를 반환
                print(f"[Tool Response]\n{result_text}")

                # 각 도구 호출 결과를 메시지에 추가합니다 (호출 개수만큼).
                messages.append({"role": "tool", "content": result_text, "name": tool_name})

        if message_obj.get("thinking"):
            print(f"\n[Agent Thinking]:\n{message_obj['thinking']}")
        print(f"\n[Agent Final Answer]:\n{message_obj['content']}")

# Jupyter Notebook 환경에서의 asyncio 실행 대응
try:
    import nest_asyncio
    nest_asyncio.apply()
except ImportError:
    pass

try:
    await run_agent("딥러닝 관련 도서를 데이터베이스에서 추천해줘.")
except Exception as e:
    print(f"\n에이전트 구동에 실패했습니다: {e}")
    print("Ollama 서버가 실행 중이고 qwen3:latest 모델이 당겨져(pull) 있는지, `fastmcp`가 설치되어 있는지 확인하세요.")