Model Context Protocol（MCP）完全ガイド：AIツール標準化とClaude統合まで

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/tmp/example.txt"
    }
  }
}

[Host: Claude Desktop]
    └── [Client] ──── stdio/SSE ──── [MCP Server: filesystem]
    └── [Client] ──── stdio/SSE ──── [MCP Server: github]
    └── [Client] ──── stdio/SSE ──── [MCP Server: database]

@mcp.prompt()
def code_review_prompt(language: str, code: str) -> str:
    return f"{language}のコードをレビューして改善点を教えてください:\n\n{code}"

pip install fastmcp

from fastmcp import FastMCP

mcp = FastMCP("filesystem-server")

@mcp.tool()
def read_file(path: str) -> str:
    """ファイルの内容を読み取って返します。"""
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

@mcp.tool()
def write_file(path: str, content: str) -> str:
    """ファイルに内容を書き込みます。"""
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"ファイル保存完了: {path}"

@mcp.resource("file://{path}")
def get_file_resource(path: str) -> str:
    """URIを通じてファイルリソースを提供します。"""
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

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

import os
from pathlib import Path
from fastmcp import FastMCP

mcp = FastMCP("filesystem-server")

@mcp.tool()
def list_directory(path: str = ".") -> list[str]:
    """ディレクトリのファイル一覧を返します。"""
    p = Path(path)
    if not p.is_dir():
        raise ValueError(f"{path} はディレクトリではありません")
    return [str(item) for item in p.iterdir()]

@mcp.tool()
def search_files(directory: str, pattern: str) -> list[str]:
    """ディレクトリ内でパターンに一致するファイルを検索します。"""
    p = Path(directory)
    return [str(f) for f in p.rglob(pattern)]

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

import { McpServer, ResourceTemplate } from '@modelcontextprotocol/sdk/server/mcp.js'
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'
import { z } from 'zod'

const server = new McpServer({
  name: 'database-server',
  version: '1.0.0',
})

// リソース: DBレコードをURIで公開
server.resource(
  'user',
  new ResourceTemplate('db://users/{id}', { list: undefined }),
  async (uri, { id }) => {
    const user = await fetchUserById(id)
    return {
      contents: [
        {
          uri: uri.href,
          text: JSON.stringify(user, null, 2),
          mimeType: 'application/json',
        },
      ],
    }
  }
)

// ツール: SQLクエリを実行
server.tool('query_db', { sql: z.string().describe('実行するSQLクエリ') }, async ({ sql }) => {
  const results = await executeQuery(sql)
  return {
    content: [
      {
        type: 'text',
        text: JSON.stringify(results, null, 2),
      },
    ],
  }
})

async function main() {
  const transport = new StdioServerTransport()
  await server.connect(transport)
}

main()

# SSEサーバーの起動例
if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

{
  "mcpServers": {
    "filesystem": {
      "command": "python",
      "args": ["/path/to/filesystem_server.py"],
      "env": {
        "ALLOWED_DIRS": "/home/user/documents,/tmp"
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_your_token_here"
      }
    },
    "sqlite": {
      "command": "uvx",
      "args": ["mcp-server-sqlite", "--db-path", "/path/to/database.db"]
    }
  }
}

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

async def main():
    server_params = StdioServerParameters(
        command="python",
        args=["filesystem_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初期化
            await session.initialize()

            # 利用可能なツール一覧
            tools = await session.list_tools()
            print("ツール一覧:", [t.name for t in tools.tools])

            # ツールの呼び出し
            result = await session.call_tool(
                "read_file",
                arguments={"path": "/tmp/test.txt"}
            )
            print("結果:", result.content[0].text)

asyncio.run(main())

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio

async def create_mcp_agent():
    model = ChatAnthropic(model="claude-3-5-sonnet-20241022")

    server_params = StdioServerParameters(
        command="python",
        args=["filesystem_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # MCPツールをLangChainツールに変換
            tools = await load_mcp_tools(session)

            # ReActエージェントを作成
            agent = create_react_agent(model, tools)

            result = await agent.ainvoke({
                "messages": [{"role": "user", "content": "/tmp ディレクトリのファイル一覧を表示してください"}]
            })
            return result

asyncio.run(create_mcp_agent())

from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

mcp = FastMCP("safe-server")

DANGEROUS_PATHS = ["/etc", "/sys", "/proc", "/root"]

@mcp.tool()
def delete_file(path: str, confirmed: bool = False) -> str:
    """
    ファイルを削除します。削除前に必ず confirmed=True を明示してください。

    Args:
        path: 削除するファイルのパス
        confirmed: 削除意思の明示的な確認（デフォルト: False）
    """
    import os
    from pathlib import Path

    abs_path = str(Path(path).resolve())

    # 危険パスのブロック
    for danger in DANGEROUS_PATHS:
        if abs_path.startswith(danger):
            raise ToolError(f"{danger} 配下のパスは削除できません")

    # 確認なしで呼び出された場合、ユーザーに確認を要求
    if not confirmed:
        return (
            f"警告: {abs_path} を完全に削除しようとしています。"
            f"confirmed=True を設定して再度呼び出すと削除されます。"
        )

    os.remove(abs_path)
    return f"削除完了: {abs_path}"

import os
from pathlib import Path

ALLOWED_DIRS = os.getenv("ALLOWED_DIRS", "/tmp").split(",")

def is_path_allowed(path: str) -> bool:
    abs_path = Path(path).resolve()
    return any(
        str(abs_path).startswith(allowed.strip())
        for allowed in ALLOWED_DIRS
    )

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .
# rootではなく専用ユーザーで実行
RUN useradd -m mcpuser
USER mcpuser
CMD ["python", "server.py"]

from pydantic import BaseModel, validator

class FileReadParams(BaseModel):
    path: str

    @validator("path")
    def validate_path(cls, v):
        p = Path(v).resolve()
        if ".." in str(p):
            raise ValueError("パストラバーサルの試みを検出しました")
        if not is_path_allowed(str(p)):
            raise ValueError("許可されていないパスです")
        return str(p)