AIエージェント開発完全ガイド2025：Tool Calling、ReAct、Multi-Agent、MCPまで

┌─────────────────────────────────────────┐
│              AI Agent                    │
│                                          │
│  ┌──────────┐  ┌──────────┐  ┌────────┐ │
│  │   LLM    │  │  Tools   │  │ Memory │ │
│  │ (Brain)  │  │(Actions) │  │(State) │ │
│  └────┬─────┘  └────┬─────┘  └───┬────┘ │
│       │              │            │      │
│       └──────┬───────┴────────────┘      │
│              │                           │
│       ┌──────▼──────┐                    │
│       │  Planning   │                    │
│       │  (Strategy) │                    │
│       └─────────────┘                    │
└─────────────────────────────────────────┘

レベル0: Chatbot
  ユーザー → LLM → 応答
  （単純な会話、外部ツールなし）

レベル1: RAG Pipeline
  ユーザー → 検索 → LLM + 文書 → 応答
  （検索増強生成、固定パイプライン）

レベル2: Tool-Using Agent
  ユーザー → LLM → [ツール選択 → 実行 → 結果観察] → 応答
  （自律的ツール選択、単一ループ）

レベル3: Multi-Step Agent
  ユーザー → LLM → [計画 → ツール1 → 観察 → ツール2 → ... → 最終] → 応答
  （多段階計画および実行）

レベル4: Multi-Agent System
  ユーザー → オーケストレータ → [Agent1, Agent2, Agent3] → 統合応答
  （複数の専門エージェント協業）

          ┌─────────────┐
          │ Multi-Agent │  ← 複雑な協業タスク
          │  Systems    │
         ┌┴─────────────┴┐
         │  Multi-Step   │  ← 多段階計画 + 実行
         │   Planning    │
        ┌┴───────────────┴┐
        │   Tool Calling  │  ← 外部ツール使用
        │   + Actions     │
       ┌┴─────────────────┴┐
       │  Reasoning (CoT)  │  ← 思考チェーン
       │                   │
      ┌┴───────────────────┴┐
      │    Text Generation  │  ← 基本テキスト生成
      └─────────────────────┘

ユーザー: 「ソウルの天気を教えて」
        │
        ▼
┌─────────────────┐
│   LLM 推論      │  「天気の質問なのでget_weatherを呼ぶべきだ」
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Tool Call 生成  │  get_weather(location="Seoul")
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  関数実行        │  → Weather API 呼び出し → 結果返却
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  LLM 応答生成   │  「ソウルの現在の気温は15度で晴れです」
└─────────────────┘

import openai

client = openai.OpenAI()

# ツール定義
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a given location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City name (e.g., Seoul, Tokyo)"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

# 最初の呼び出し：LLMがツール呼び出しを決定
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "ソウルの天気は？"}
    ],
    tools=tools,
    tool_choice="auto"  # auto / none / required / specific
)

# ツール呼び出しの確認と実行
message = response.choices[0].message
if message.tool_calls:
    tool_call = message.tool_calls[0]
    # 実際の関数実行
    result = get_weather(
        location=json.loads(tool_call.function.arguments)["location"]
    )

    # 結果をLLMに再度渡す
    final_response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "user", "content": "ソウルの天気は？"},
            message,  # assistantのtool_callメッセージ
            {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            }
        ],
        tools=tools
    )

import anthropic

client = anthropic.Anthropic()

# ツール定義
tools = [
    {
        "name": "get_weather",
        "description": "Get current weather for a given location",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City name (e.g., Seoul, Tokyo)"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit"
                }
            },
            "required": ["location"]
        }
    }
]

# 最初の呼び出し
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=tools,
    messages=[
        {"role": "user", "content": "ソウルの天気を教えて"}
    ]
)

# stop_reasonが"tool_use"の場合の処理
if response.stop_reason == "tool_use":
    tool_block = next(
        block for block in response.content
        if block.type == "tool_use"
    )

    # ツール実行
    result = get_weather(**tool_block.input)

    # 結果を再度渡す
    final_response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        tools=tools,
        messages=[
            {"role": "user", "content": "ソウルの天気を教えて"},
            {"role": "assistant", "content": response.content},
            {
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": tool_block.id,
                        "content": json.dumps(result)
                    }
                ]
            }
        ]
    )

import google.generativeai as genai

# 関数宣言
get_weather_func = genai.protos.FunctionDeclaration(
    name="get_weather",
    description="Get current weather for a location",
    parameters=genai.protos.Schema(
        type=genai.protos.Type.OBJECT,
        properties={
            "location": genai.protos.Schema(
                type=genai.protos.Type.STRING,
                description="City name"
            ),
        },
        required=["location"]
    )
)

tool = genai.protos.Tool(
    function_declarations=[get_weather_func]
)

model = genai.GenerativeModel(
    model_name="gemini-2.0-flash",
    tools=[tool]
)

chat = model.start_chat()
response = chat.send_message("ソウルの天気は？")

# function_call処理
if response.candidates[0].content.parts[0].function_call:
    fc = response.candidates[0].content.parts[0].function_call
    result = get_weather(location=fc.args["location"])

    response = chat.send_message(
        genai.protos.Content(
            parts=[genai.protos.Part(
                function_response=genai.protos.FunctionResponse(
                    name="get_weather",
                    response={"result": result}
                )
            )]
        )
    )

{
  "name": "search_products",
  "description": "Search for products in the e-commerce catalog. Returns a list of matching products with price, rating, and availability. Use when the user wants to find or browse products.",
  "parameters": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "Search query text. Can be product name, category, or keywords."
      },
      "category": {
        "type": "string",
        "enum": ["electronics", "clothing", "home", "books", "toys"],
        "description": "Product category filter. Optional."
      },
      "min_price": {
        "type": "number",
        "description": "Minimum price in USD. Optional."
      },
      "max_price": {
        "type": "number",
        "description": "Maximum price in USD. Optional."
      },
      "sort_by": {
        "type": "string",
        "enum": ["relevance", "price_low", "price_high", "rating", "newest"],
        "description": "Sort order for results. Defaults to relevance."
      },
      "limit": {
        "type": "integer",
        "description": "Max number of results to return (1-50). Defaults to 10."
      }
    },
    "required": ["query"]
  }
}

# OpenAI - 並列呼び出しをネイティブサポート
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "ソウルと東京の天気を比較して"}
    ],
    tools=tools,
    parallel_tool_calls=True  # デフォルト True
)

# レスポンスに複数のtool_callsが含まれる
for tool_call in message.tool_calls:
    pass

# 並列実行して全結果を渡す
import asyncio

async def execute_parallel_tools(tool_calls):
    tasks = []
    for tc in tool_calls:
        args = json.loads(tc.function.arguments)
        tasks.append(execute_tool(tc.function.name, args))
    return await asyncio.gather(*tasks)

# Auto: LLMがツール使用有無を自律的に判断
tool_choice = "auto"

# Required: 必ず1つ以上のツールを使用
tool_choice = "required"

# None: ツール使用禁止（テキスト応答のみ）
tool_choice = "none"

# 特定ツールの強制使用
tool_choice = {"type": "function", "function": {"name": "get_weather"}}

# Anthropicの場合
tool_choice = {"type": "auto"}       # 自動
tool_choice = {"type": "any"}        # 必ず1つ
tool_choice = {"type": "tool", "name": "get_weather"}  # 特定ツール強制

# OpenAI Streaming Tool Calls
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    stream=True
)

tool_calls_buffer = {}
for chunk in stream:
    delta = chunk.choices[0].delta

    if delta.tool_calls:
        for tc in delta.tool_calls:
            idx = tc.index
            if idx not in tool_calls_buffer:
                tool_calls_buffer[idx] = {
                    "id": tc.id,
                    "function": {"name": "", "arguments": ""}
                }
            if tc.function.name:
                tool_calls_buffer[idx]["function"]["name"] += tc.function.name
            if tc.function.arguments:
                tool_calls_buffer[idx]["function"]["arguments"] += tc.function.arguments

# Anthropic Streaming Tool Use
with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=tools,
    messages=messages,
) as stream:
    for event in stream:
        if event.type == "content_block_start":
            if event.content_block.type == "tool_use":
                print(f"Tool: {event.content_block.name}")
        elif event.type == "content_block_delta":
            if event.delta.type == "input_json_delta":
                print(event.delta.partial_json)

# BAD: 曖昧な説明
{
    "name": "search",
    "description": "Search for things"  # 曖昧すぎ
}

# GOOD: 明確で具体的な説明
{
    "name": "search_knowledge_base",
    "description": "Search the internal knowledge base for technical documentation, API references, and troubleshooting guides. Use this when the user asks about product features, API usage, error codes, or configuration options. Do NOT use for general web search or current events.",
}

# BEST: 例と使用条件を含む
{
    "name": "search_knowledge_base",
    "description": """Search the internal knowledge base for technical documentation.

USE WHEN:
- User asks about product features or capabilities
- User needs API reference or code examples
- User encounters error codes or needs troubleshooting

DO NOT USE WHEN:
- User asks about pricing (use get_pricing instead)
- User asks about their account (use get_account_info instead)

EXAMPLES:
- "How do I authenticate?" -> query="API authentication"
- "What does error 429 mean?" -> query="error 429 rate limit"
""",
}

# BAD
"location": {
    "type": "string",
    "description": "location"
}

# GOOD
"location": {
    "type": "string",
    "description": "City name in English. Use the most common name. Examples: 'Seoul', 'New York', 'Tokyo'. Use full names (e.g., 'Los Angeles' not 'LA')."
}

# BAD: 細分化されすぎたツール（LLMが混乱）
tools = [
    "get_user_name",
    "get_user_email",
    "get_user_phone",
    "get_user_address",
    # ... 20個以上のツール
]

# GOOD: 関連ツールを1つに統合
tools = [
    {
        "name": "get_user_info",
        "description": "Get user information. Specify which fields you need.",
        "parameters": {
            "type": "object",
            "properties": {
                "user_id": {"type": "string"},
                "fields": {
                    "type": "array",
                    "items": {
                        "type": "string",
                        "enum": ["name", "email", "phone", "address", "preferences"]
                    },
                    "description": "List of fields to retrieve."
                }
            }
        }
    }
]

system_prompt = """You are a helpful assistant with access to tools.

Example 1:
User: "What's the weather in Seoul?"
Action: Call get_weather with location="Seoul", unit="celsius"

Example 2:
User: "Find me a good Italian restaurant"
Action: Call search_restaurants with cuisine="italian", sort_by="rating"

Example 3:
User: "What's 2+2?"
Action: Do NOT use any tools. Respond directly: "2+2 = 4"
"""

# BAD: 深くネストされたスキーマ
{
    "type": "object",
    "properties": {
        "filter": {
            "type": "object",
            "properties": {
                "conditions": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "field": {"type": "string"},
                            "operator": {"type": "string"},
                            "value": {"type": "object"}  # 3段階ネスト
                        }
                    }
                }
            }
        }
    }
}

# GOOD: フラットなスキーマに変換
{
    "type": "object",
    "properties": {
        "filter_field": {"type": "string", "description": "Field to filter on"},
        "filter_operator": {"type": "string", "enum": ["eq", "gt", "lt", "contains"]},
        "filter_value": {"type": "string", "description": "Filter value as string"}
    }
}

import tenacity

class ToolExecutor:
    def __init__(self, client, tools, model="gpt-4o"):
        self.client = client
        self.tools = tools
        self.model = model

    @tenacity.retry(
        stop=tenacity.stop_after_attempt(3),
        wait=tenacity.wait_exponential(min=1, max=10),
        retry=tenacity.retry_if_exception_type(ToolExecutionError)
    )
    async def execute_tool(self, name, arguments):
        try:
            result = await self.tool_registry[name](**arguments)
            return {"status": "success", "result": result}
        except ValidationError as e:
            return {
                "status": "error",
                "error": f"Invalid parameters: {str(e)}",
                "hint": "Please check the parameter types and try again"
            }
        except RateLimitError:
            raise ToolExecutionError("Rate limited, will retry")
        except Exception as e:
            return {
                "status": "error",
                "error": f"Tool execution failed: {str(e)}"
            }

    async def run_agent_loop(self, messages, max_iterations=10):
        for i in range(max_iterations):
            response = await self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=self.tools
            )

            if response.choices[0].finish_reason == "stop":
                return response.choices[0].message.content

            for tool_call in response.choices[0].message.tool_calls:
                result = await self.execute_tool(
                    tool_call.function.name,
                    json.loads(tool_call.function.arguments)
                )
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result)
                })

        return "Max iterations reached"

import hashlib

class CachedToolExecutor:
    def __init__(self):
        self.cache = {}
        self.cache_ttl = 300  # 5分

    def _cache_key(self, tool_name, arguments):
        arg_str = json.dumps(arguments, sort_keys=True)
        return hashlib.md5(
            f"{tool_name}:{arg_str}".encode()
        ).hexdigest()

    async def execute_with_cache(self, tool_name, arguments):
        CACHEABLE_TOOLS = {"get_weather", "search_products", "get_stock_price"}

        if tool_name not in CACHEABLE_TOOLS:
            return await self.execute_tool(tool_name, arguments)

        key = self._cache_key(tool_name, arguments)
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["timestamp"] < self.cache_ttl:
                return entry["result"]

        result = await self.execute_tool(tool_name, arguments)
        self.cache[key] = {
            "result": result,
            "timestamp": time.time()
        }
        return result

import asyncio

async def optimized_tool_execution(tool_calls):
    """並列実行 + タイムアウト管理"""

    independent_calls = group_independent_calls(tool_calls)

    results = {}
    for group in independent_calls:
        group_tasks = [
            asyncio.wait_for(
                execute_tool(tc.name, tc.args),
                timeout=30.0  # 30秒タイムアウト
            )
            for tc in group
        ]
        group_results = await asyncio.gather(
            *group_tasks, return_exceptions=True
        )
        for tc, result in zip(group, group_results):
            if isinstance(result, Exception):
                results[tc.id] = {"error": str(result)}
            else:
                results[tc.id] = result

    return results

# UnslothによるTool Callingファインチューニング
from unsloth import FastLanguageModel

model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="unsloth/Llama-3.1-8B-Instruct",
    max_seq_length=4096,
    load_in_4bit=True,
)

model = FastLanguageModel.get_peft_model(
    model,
    r=16,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                     "gate_proj", "up_proj", "down_proj"],
    lora_alpha=16,
    lora_dropout=0,
    bias="none",
)

# Tool Calling学習データ形式
training_data = [
    {
        "messages": [
            {"role": "system", "content": "You are an assistant with tools..."},
            {"role": "user", "content": "What's the weather in Seoul?"},
            {"role": "assistant", "content": None, "tool_calls": [
                {
                    "type": "function",
                    "function": {
                        "name": "get_weather",
                        "arguments": '{"location": "Seoul", "unit": "celsius"}'
                    }
                }
            ]},
            {"role": "tool", "content": '{"temp": 15, "condition": "sunny"}'},
            {"role": "assistant", "content": "ソウルの現在の天気は15度で晴れです。"}
        ]
    }
]

質問: 「2025年のソウルの人口は東京の何%か？」

Thought 1: ソウルと東京の2025年の人口をそれぞれ調べる必要がある。
Action 1: search("2025年 ソウル 人口")
Observation 1: ソウルの2025年の人口は約940万人です。

Thought 2: 次に東京の人口を調べなければならない。
Action 2: search("2025年 東京 人口")
Observation 2: 東京の2025年の人口は約1,396万人です。

Thought 3: 比率を計算できる。940/1396 * 100
Action 3: calculator("940 / 1396 * 100")
Observation 3: 67.34

Thought 4: 計算完了。答えをまとめよう。
Final Answer: 2025年のソウルの人口（約940万）は東京（約1,396万）の約67.3%です。

import re
import json
from openai import OpenAI

class ReActAgent:
    def __init__(self, tools: dict, model: str = "gpt-4o"):
        self.client = OpenAI()
        self.tools = tools
        self.model = model
        self.max_steps = 10

    def _build_system_prompt(self):
        tool_descriptions = "\n".join([
            f"- {name}: {func.__doc__}"
            for name, func in self.tools.items()
        ])

        return f"""You are an AI assistant that uses a Thought-Action-Observation loop.

Available tools:
{tool_descriptions}

For each step, output EXACTLY ONE of:
1. Thought: [your reasoning]
2. Action: tool_name(arg1="value1", arg2="value2")
3. Final Answer: [your complete answer]

Rules:
- Always start with a Thought
- After each Action, wait for an Observation
- When you have enough information, give a Final Answer
"""

    def run(self, question: str) -> str:
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": question}
        ]

        for step in range(self.max_steps):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0
            )
            output = response.choices[0].message.content
            messages.append({"role": "assistant", "content": output})

            if "Final Answer:" in output:
                return output.split("Final Answer:")[-1].strip()

            action_match = re.search(
                r'Action:\s*(\w+)\((.*?)\)', output, re.DOTALL
            )
            if action_match:
                tool_name = action_match.group(1)
                args_str = action_match.group(2)
                args = self._parse_args(args_str)

                if tool_name in self.tools:
                    try:
                        observation = self.tools[tool_name](**args)
                    except Exception as e:
                        observation = f"Error: {str(e)}"
                else:
                    observation = f"Error: Unknown tool '{tool_name}'"

                messages.append({
                    "role": "user",
                    "content": f"Observation: {observation}"
                })

        return "Max steps reached without final answer"

    def _parse_args(self, args_str: str) -> dict:
        args = {}
        for match in re.finditer(r'(\w+)="([^"]*)"', args_str):
            args[match.group(1)] = match.group(2)
        return args

# LangGraph - グラフベースエージェント
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    return web_search(query)

class AgentState(TypedDict):
    messages: list
    next_step: str

def should_continue(state):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return "end"

workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", execute_tools)
workflow.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    "end": END,
})
workflow.add_edge("tools", "agent")
workflow.set_entry_point("agent")

app = workflow.compile()

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role='Senior Research Analyst',
    goal='Find the latest trends in AI agent development',
    backstory='Expert researcher with deep knowledge of AI/ML',
    tools=[search_tool, web_scraper],
    llm='gpt-4o',
    verbose=True
)

writer = Agent(
    role='Tech Content Writer',
    goal='Create engaging technical blog posts',
    backstory='Experienced tech writer specializing in AI topics',
    tools=[write_tool],
    llm='gpt-4o',
    verbose=True
)

editor = Agent(
    role='Editor',
    goal='Ensure content quality and accuracy',
    backstory='Senior editor with technical background',
    llm='gpt-4o',
    verbose=True
)

research_task = Task(
    description='Research the latest AI agent frameworks',
    expected_output='Comprehensive research report',
    agent=researcher
)

writing_task = Task(
    description='Write a blog post based on findings',
    expected_output='Well-structured blog post',
    agent=writer,
    context=[research_task]
)

editing_task = Task(
    description='Review and improve the blog post',
    expected_output='Publication-ready blog post',
    agent=editor,
    context=[writing_task]
)

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

assistant = AssistantAgent(
    name="AI_Assistant",
    llm_config={"model": "gpt-4o"},
    system_message="You are a helpful AI assistant."
)

coder = AssistantAgent(
    name="Coder",
    llm_config={"model": "gpt-4o"},
    system_message="You are an expert Python programmer."
)

user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "workspace"},
)

group_chat = GroupChat(
    agents=[user_proxy, assistant, coder],
    messages=[],
    max_round=10
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config={"model": "gpt-4o"}
)

user_proxy.initiate_chat(
    manager,
    message="Create a Python script that analyzes CSV data"
)

                ┌──────────────┐
                │  Supervisor  │
                │   Agent      │
                └──────┬───────┘
                       │
          ┌────────────┼────────────┐
          │            │            │
    ┌─────▼─────┐ ┌───▼────┐ ┌────▼────┐
    │ Research  │ │ Writer │ │ Analyst │
    │  Agent    │ │ Agent  │ │  Agent  │
    └───────────┘ └────────┘ └─────────┘

    ┌──────────┐     ┌──────────┐
    │ Agent A  │◄───►│ Agent B  │
    └────┬─────┘     └─────┬────┘
         │                  │
         └──────┬───────────┘
                │
         ┌──────▼─────┐
         │  Agent C   │
         └────────────┘

              ┌────────────┐
              │   CEO      │
              │  Agent     │
              └─────┬──────┘
                    │
         ┌──────────┼──────────┐
         │                     │
   ┌─────▼──────┐       ┌─────▼──────┐
   │ Tech Lead  │       │  PM Lead   │
   │  Agent     │       │   Agent    │
   └─────┬──────┘       └─────┬──────┘
         │                     │
    ┌────┼────┐          ┌─────┼─────┐
    │         │          │           │
┌───▼──┐ ┌───▼──┐  ┌────▼──┐  ┌────▼──┐
│Dev 1 │ │Dev 2 │  │Design │  │  QA   │
│Agent │ │Agent │  │Agent  │  │Agent  │
└──────┘ └──────┘  └───────┘  └───────┘

Before MCP:
┌─────────────┐     ┌──────────────────────────┐
│  Agent App  │────►│ カスタムAPIラッパー (GitHub) │
│             │────►│ カスタムAPIラッパー (Slack)  │
│             │────►│ カスタムAPIラッパー (DB)     │
│             │────►│ カスタムAPIラッパー (ファイル)│
└─────────────┘     └──────────────────────────┘
  → ツールごとに異なるインターフェース、重複コード

After MCP:
┌─────────────┐     ┌──────────────┐     ┌──────────┐
│  Agent App  │────►│  MCP Client  │────►│ GitHub   │
│  (MCP Host) │     │              │────►│ Slack    │
│             │     │  標準         │────►│ Database │
│             │     │  プロトコル    │────►│ Files    │
└─────────────┘     └──────────────┘     └──────────┘
  → 統一インターフェース、再利用可能、拡張容易

┌────────────────────────────────────────────┐
│                MCP Host                     │
│  (Claude Desktop, IDE, Agent Application)   │
│                                             │
│  ┌─────────────┐  ┌─────────────┐          │
│  │ MCP Client  │  │ MCP Client  │  ...      │
│  │ (GitHub)    │  │ (Database)  │          │
│  └──────┬──────┘  └──────┬──────┘          │
└─────────┼────────────────┼─────────────────┘
          │  stdio/SSE     │  stdio/SSE
┌─────────▼──────┐  ┌─────▼────────┐
│  MCP Server    │  │  MCP Server  │
│  (GitHub)      │  │  (Database)  │
│  Resources     │  │  Resources   │
│  Tools         │  │  Tools       │
│  Prompts       │  │  Prompts     │
└────────────────┘  └──────────────┘

from mcp.server import Server
from mcp.types import Tool, TextContent
import json

server = Server("todo-server")
todos = []

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="add_todo",
            description="Add a new todo item",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {"type": "string", "description": "Todo title"},
                    "priority": {
                        "type": "string",
                        "enum": ["low", "medium", "high"]
                    }
                },
                "required": ["title"]
            }
        ),
        Tool(
            name="list_todos",
            description="List all todo items",
            inputSchema={"type": "object", "properties": {}}
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "add_todo":
        todo = {
            "title": arguments["title"],
            "priority": arguments.get("priority", "medium"),
            "completed": False
        }
        todos.append(todo)
        return [TextContent(
            type="text",
            text=f"Added: {todo['title']} ({todo['priority']})"
        )]
    elif name == "list_todos":
        if not todos:
            return [TextContent(type="text", text="No todos yet!")]
        result = "\n".join([
            f"{'[x]' if t['completed'] else '[ ]'} {i}. {t['title']}"
            for i, t in enumerate(todos)
        ])
        return [TextContent(type="text", text=result)]

if __name__ == "__main__":
    import asyncio
    from mcp.server.stdio import stdio_server

    async def main():
        async with stdio_server() as (read_stream, write_stream):
            await server.run(read_stream, write_stream)

    asyncio.run(main())

class ConversationMemory:
    def __init__(self, max_messages: int = 20):
        self.messages = []
        self.max_messages = max_messages

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        if len(self.messages) > self.max_messages:
            system = [m for m in self.messages if m["role"] == "system"]
            recent = self.messages[-self.max_messages:]
            self.messages = system + recent

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

class LongTermMemory:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = Chroma(
            collection_name="agent_memory",
            embedding_function=self.embeddings,
            persist_directory="./memory_db"
        )

    def store(self, text: str, metadata: dict = None):
        self.vectorstore.add_texts(
            texts=[text], metadatas=[metadata or {}]
        )

    def recall(self, query: str, k: int = 5) -> list:
        results = self.vectorstore.similarity_search(query, k=k)
        return [doc.page_content for doc in results]

class EpisodicMemory:
    """過去のインタラクションの成功/失敗パターンを記憶"""

    def __init__(self):
        self.episodes = []

    def record_episode(self, task, actions, outcome, success):
        self.episodes.append({
            "task": task,
            "actions": actions,
            "outcome": outcome,
            "success": success,
            "timestamp": datetime.now().isoformat()
        })

    def recall_similar(self, current_task, k=3):
        similar = find_similar_episodes(current_task, self.episodes, k)
        return similar

class WorkingMemory:
    """エージェントの現在のタスク状態を管理"""

    def __init__(self):
        self.scratchpad = {}
        self.current_plan = []
        self.completed_steps = []

    def update_plan(self, plan: list):
        self.current_plan = plan

    def mark_step_done(self, step_index: int, result: str):
        self.completed_steps.append({
            "step": self.current_plan[step_index],
            "result": result
        })

    def get_context(self) -> str:
        return f"""Current Plan: {json.dumps(self.current_plan)}
Completed Steps: {json.dumps(self.completed_steps)}
Scratchpad: {json.dumps(self.scratchpad)}"""

class SafeToolExecutor:
    ALLOWED_TOOLS = {
        "search_web", "get_weather", "search_products",
        "get_user_info", "create_ticket"
    }

    DANGEROUS_TOOLS = {
        "delete_user", "execute_sql", "send_email",
        "modify_config", "deploy_service"
    }

    def __init__(self):
        self.rate_limiter = RateLimiter(
            max_calls=100,
            time_window=60  # 毎分100回
        )

    async def execute(self, tool_name, args, user_context):
        # 1. 許可ツール確認
        if tool_name not in self.ALLOWED_TOOLS:
            if tool_name in self.DANGEROUS_TOOLS:
                return await self.request_human_approval(
                    tool_name, args, user_context
                )
            raise ToolNotAllowedError(f"'{tool_name}' is not whitelisted")

        # 2. レートリミット確認
        if not self.rate_limiter.allow(user_context.user_id):
            raise RateLimitExceededError()

        # 3. 入力検証
        self.validate_args(tool_name, args)

        # 4. 実行
        return await self.tool_registry[tool_name](**args)

# LangSmith統合
from langsmith import traceable

@traceable(name="agent_run")
async def run_agent(query: str, session_id: str):
    with trace_context(session_id=session_id):
        result = await agent.invoke(query)
    return result

# Langfuse統合
from langfuse import Langfuse

langfuse = Langfuse()

def traced_tool_call(tool_name, args):
    generation = langfuse.generation(
        name=f"tool_call_{tool_name}",
        input=args,
        model="gpt-4o"
    )
    try:
        result = execute_tool(tool_name, args)
        generation.end(output=result, status="success")
        return result
    except Exception as e:
        generation.end(output=str(e), status="error")
        raise

class CostTracker:
    # 価格テーブル（USD/1Mトークン、2025年基準）
    PRICING = {
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
        "claude-3-5-haiku": {"input": 0.80, "output": 4.00},
    }

    def __init__(self, budget_limit: float = 10.0):
        self.total_cost = 0.0
        self.budget_limit = budget_limit

    def track(self, model, input_tokens, output_tokens):
        pricing = self.PRICING.get(model, {"input": 0, "output": 0})
        cost = (
            input_tokens / 1_000_000 * pricing["input"] +
            output_tokens / 1_000_000 * pricing["output"]
        )
        self.total_cost += cost

        if self.total_cost > self.budget_limit:
            raise BudgetExceededError(
                f"Budget exceeded: ${self.total_cost:.4f}"
            )

class ResilientAgent:
    async def run_with_fallback(self, query, max_retries=3):
        for attempt in range(max_retries):
            try:
                return await self.primary_agent.run(query)
            except ToolExecutionError as e:
                logger.warning(f"Tool error (attempt {attempt+1}): {e}")
                if attempt < max_retries - 1:
                    await self.reset_tools()
                    continue
            except LLMError as e:
                logger.warning(f"LLM error: {e}, fallback使用")
                return await self.fallback_agent.run(query)
            except BudgetExceededError:
                return "予算超過。より簡単なクエリをお試しください。"

        return "複数回の試行後もリクエストを完了できませんでした。"