Client 구현

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

async def main():
    # 서버 프로세스 설정
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"],
        env={"DB_URL": "sqlite:///app.db"}  # 서버 환경변수
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 1. 연결 초기화 (initialize + initialized)
            await session.initialize()

            # 2. 사용 가능한 Tool 목록 조회
            tools_result = await session.list_tools()
            for tool in tools_result.tools:
                print(f"Tool: {tool.name} — {tool.description}")

            # 3. Tool 호출
            result = await session.call_tool(
                "web_search",
                arguments={"query": "MCP Python SDK", "max_results": 3}
            )
            if not result.isError:
                for content in result.content:
                    print(content.text)

            # 4. Resource 읽기
            resource = await session.read_resource("config://app/settings")
            for content in resource.contents:
                print(content.text)

asyncio.run(main())

from mcp.client.sse import sse_client

async def connect_remote():
    async with sse_client("http://my-mcp-server:8080/sse") as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            # 이후 사용법은 stdio와 동일
            result = await session.call_tool("query_db", {"sql": "SELECT * FROM users LIMIT 10"})
            return result

pip install langchain-mcp-adapters langgraph langchain-anthropic

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

async def build_agent():
    # 여러 MCP Server를 한번에 연결
    client = MultiServerMCPClient({
        "search": {
            "command": "python",
            "args": ["search_server.py"],
            "transport": "stdio",
        },
        "database": {
            "url": "http://db-server:8080/sse",
            "transport": "sse",
        },
    })

    async with client as mcp:
        # MCP Tool → LangChain Tool 자동 변환
        tools = await mcp.get_tools()

        model = ChatAnthropic(model="claude-opus-4-7")
        agent = create_react_agent(model, tools)

        result = await agent.ainvoke({
            "messages": [{"role": "user", "content": "최근 가입 사용자 10명을 검색해서 알려줘"}]
        })
        return result

asyncio.run(build_agent())

from mcp import McpError
from mcp.types import ErrorCode
import asyncio

async def safe_tool_call(
    session: ClientSession,
    tool_name: str,
    arguments: dict,
    retries: int = 3
) -> str | None:
    for attempt in range(retries):
        try:
            result = await session.call_tool(tool_name, arguments)

            if result.isError:
                # Tool 수준 에러 — 재시도하지 않음
                error_text = result.content[0].text if result.content else "Unknown error"
                print(f"Tool error: {error_text}")
                return None

            return result.content[0].text

        except McpError as e:
            if e.error.code == ErrorCode.INVALID_PARAMS:
                raise  # 잘못된 파라미터 — 재시도 무의미
            if attempt == retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # 지수 백오프

        except Exception as e:
            print(f"Attempt {attempt+1} failed: {e}")
            if attempt == retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)