DEVOCEAN

mkdir mcp_server
cd mcp_server
uv init --python 3.12

from mcp.server.fastmcp import FastMCP
import random

mcp = FastMCP("reco_food")
menus = {
    "중식": ["짜장면", "마라탕"],
    "일식": ["스시", "우동"],
    "한식": ["비빔밥", "김치찌개"]
}

@mcp.tool()
def recommend_food() -> str:
    """식사 메뉴를 추천하는 함수

    Returns:
        str: 오늘 메뉴
    """
    # 카테고리 무작위 선택
    category = random.choice(list(menus.keys()))

    # 해당 카테고리에서 음식 무작위 선택
    menu = random.choice(menus[category])

    return f"오늘의 메뉴는 ({category}): {menu}"


if __name__ == "__main__":
    print("Starting MCP server...")
    mcp.run()

{
  "mcpServers": {
    "reco_food": {
      "command": "/Users/victor/.local/bin/uv",
      "args": [
        "--directory",
        "/Users/victor/dev/mcp-example/mcp_server",
        "run",
        "reco_food_server.py"
      ]
    }
  }
}

import sys
import asyncio

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

async def connect_to_server(server_script_path: str):
    """Connect to an MCP server and list available tools"""
    if not server_script_path.endswith('.py'):
        raise ValueError("Server script must be a .py file")

    server_params = StdioServerParameters(
        command="python",
        args=[server_script_path],
        env=None,
    )

    async with stdio_client(server_params) as (stdio, write):
        session = ClientSession(stdio, write)
        async with session:
            await session.initialize()
            response = await session.list_tools()
            tools = response.tools
            print("\n✅ Connected to server with tools:", [tool.name for tool in tools])


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python list_tools.py <path_to_server_script>")
        sys.exit(1)

    asyncio.run(connect_to_server(sys.argv[1]))

uv run list_tools.py ../mcp_server/reco_food_server.py

[05/30/25 11:50:22] INFO     Processing request of type      server.py:551
                             ListToolsRequest                             

✅ Connected to server with tools: ['recommend_food']

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

async def main():
    client = MultiServerMCPClient(
        {
            "reco_food": {
                "command": "uv",
                "args": ["--directory", "/Users/victor/dev/mcp/mcp-example/mcp_server", "run", "reco_food_server.py"],  # 실제 경로로 대체하세요
                "transport": "stdio",
            }
        }
    )
    
    tools = await client.get_tools()

    # openai API key가 필요하며, openai 패키지가 설치되어 있어야 합니다
    agent = create_react_agent("openai:gpt-4.1", tools)

    math_response = await agent.ainvoke({"messages": "오늘 점심 메뉴 추천해줘"})
    print("Math Response:", math_response)

# asyncio 루프 실행
if __name__ == "__main__":
    asyncio.run(main())

Math Response: {'messages': [HumanMessage(content='오늘 점심 메뉴 추천해줘', additional_kwargs={}, response_metadata={}, id='a2aec9b0-dac9-4ec7-908a-504419fd23c6'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_xEOFbXG9yk1x5DNr3RgwCxgb', 'function': {'arguments': '{}', 'name': 'recommend_food'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 11, 'prompt_tokens': 54, 'total_tokens': 65, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4.1-2025-04-14', 'system_fingerprint': 'fp_51e1070cf2', 'id': 'chatcmpl-BclCe00OQcs4kWBaV8ac5bWb3q5Fq', 'service_tier': 'default', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run--7f6b66d1-7fec-4d53-b589-59a05d023bb9-0', tool_calls=[{'name': 'recommend_food', 'args': {}, 'id': 'call_xEOFbXG9yk1x5DNr3RgwCxgb', 'type': 'tool_call'}], usage_metadata={'input_tokens': 54, 'output_tokens': 11, 'total_tokens': 65, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}), ToolMessage(content='오늘의 메뉴는 (중식): 짜장면', name='recommend_food', id='5aa1abc6-88f7-4f3b-8a23-bc0aa34f918b', tool_call_id='call_xEOFbXG9yk1x5DNr3RgwCxgb'), AIMessage(content='오늘의 점심 메뉴로 짜장면을 추천합니다! 맛있게 드세요.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 21, 'prompt_tokens': 85, 'total_tokens': 106, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4.1-2025-04-14', 'system_fingerprint': 'fp_51e1070cf2', 'id': 'chatcmpl-BclChShCorugWoZ1etFmoDXp9WnsR', 'service_tier': 'default', 'finish_reason': 'stop', 'logprobs': None}, id='run--e36ba24f-d4b4-4f0b-8643-832e0af70bd9-0', usage_metadata={'input_tokens': 85, 'output_tokens': 21, 'total_tokens': 106, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})]}

import streamlit as st
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_core.messages.ai import AIMessageChunk

st.title("ChatGPT-like Clone with MCP (Async)")

# MCP 클라이언트 설정
client = MultiServerMCPClient(
    {
        "reco_food": {
            "command": "uv",
            "args": [
                "--directory",
                "/Users/victor/dev/mcp/mcp-example/mcp_server",
                "run",
                "reco_food_server.py"
            ],
            "transport": "stdio",
        },
        "phonebook": {
            "command": "uv",
            "args": [
                "--directory",
                "/Users/victor/dev/mcp/mcp-example/mcp_server",
                "run",
                "phonebook_server.py"
            ],
            "transport": "stdio",
        }
    }
)

# MCP 도구 비동기 로딩
async def init_agent():
    tools = await client.get_tools()
    agent = create_react_agent(
        "openai:gpt-4o-mini",
        tools=tools,
        prompt="You are a helpful assistant."
    )
    return agent

# 세션 상태 초기화
if "messages" not in st.session_state:
    st.session_state.messages = []

if "agent" not in st.session_state:
    st.session_state.agent = asyncio.run(init_agent())

# LangGraph agent 호출 비동기 함수
async def call_agent_async(agent, messages):
    return await agent.ainvoke({"messages": messages})

# 이전 대화 표시
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# 사용자 입력 처리
if prompt := st.chat_input("What is up?"):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        response = asyncio.run(call_agent_async(st.session_state.agent, st.session_state.messages))
        new_messages = response["messages"]
        
        # 마지막 AI 메시지 (최종 사용자 응답용)
        last_ai_index = next(
            (i for i in reversed(range(len(new_messages))) if new_messages[i].type == "ai" and new_messages[i].content),
            None
        )

        if last_ai_index is not None:
            last_ai_msg = new_messages[last_ai_index]

            # 직전 tool call AIMessage 확인
            tool_call_msg = None
            for i in range(last_ai_index - 1, -1, -1):
                msg = new_messages[i]
                if msg.type == "ai" and msg.additional_kwargs.get("tool_calls"):
                    tool_call_msg = msg
                    break
            
            # tool_calls 출력
            if tool_call_msg:
                st.markdown("**[Tool 호출됨]**")
                for call in tool_call_msg.additional_kwargs["tool_calls"]:
                    function_name = call["function"]["name"]
                    call_id = call["id"]

                    st.markdown(f"- `{function_name}` 함수 호출됨")

                    # tool 응답 출력
                    tool_result_msg = next(
                        (m for m in new_messages if m.type == "tool" and m.tool_call_id == call_id),
                        None
                    )
                    if tool_result_msg:
                        st.markdown(f"**[Tool 응답 결과]**\n- `{function_name}` 결과: {tool_result_msg.content}")

            # 마지막 AI 응답 출력
            st.markdown(last_ai_msg.content)

            # 세션 상태에 저장
            st.session_state.messages.append({
                "role": "assistant",
                "content": last_ai_msg.content
            })

uv run streamlit run main.py

from fastapi import FastAPI
from fastapi.responses import JSONResponse
import csv

app = FastAPI()

@app.get("/phonebook")
def read_csv_data():
    data = []
    try:
        with open("phonebook.csv", newline='', encoding='utf-8') as csvfile:
            reader = csv.DictReader(csvfile)
            for row in reader:
                data.append(row)
    except FileNotFoundError:
        return JSONResponse(status_code=404, content={"message": "CSV file not found"})
    
    return data

from mcp.server.fastmcp import FastMCP
import requests

mcp = FastMCP("phonebook")

@mcp.tool()
def fetch_phonebook():
    """phoneboo에서 연락처를 가져오는 함수

    Returns:
        json
    """
    url = "http://127.0.0.1:8000/phonebook"
    try:
        response = requests.get(url)
        response.raise_for_status()  # HTTP 오류 발생 시 예외 발생
        data = response.json()
        return data
    except requests.exceptions.RequestException as e:
        print(f"Error fetching data: {e}")
        return None
    
if __name__ == "__main__":
    print("Starting MCP server...")
    mcp.run()