mcp lesson

mcp-tools/solution-files/product_assistant_mcp.py

@@ -0,0 +1,95 @@
+from openai import OpenAI
+import json
+import os
+from product_server import mcp
+
+client = OpenAI(
+    api_key=os.environ.get("TOGETHER_API_KEY"),
+    base_url="https://api.together.xyz/v1"
+)
+
+def get_openai_tools(mcp_server):
+    """Convert MCP tools to OpenAI tools format."""
+    tools = []
+
+    for tool in mcp_server._tool_manager._tools.values():
+        tools.append({
+            "type": "function",
+            "function": {
+                "name": tool.name,
+                "description": tool.description,
+                "strict": True,
+                "parameters": tool.parameters
+            }
+        })
+
+    return tools
+
+def get_tool_functions(mcp_server):
+    """Get a mapping of tool names to their functions."""
+    return {name: tool.fn for name, tool in mcp_server._tool_manager._tools.items()}
+
+def run_agent(user_message: str, mcp_server, max_iterations: int = 10) -> str:
+    """Run the agentic loop using tools from an MCP server."""
+
+    # Convert MCP tools to OpenAI format
+    tools = get_openai_tools(mcp_server)
+    tool_functions = get_tool_functions(mcp_server)
+
+    messages = [
+        {
+            "role": "system",
+            "content": (
+                "You are a product assistant for GlobalJava Roasters. "
+                "Always use the available tools to look up current product information and pricing. "
+                "Do not rely on general knowledge about coffee."
+            )
+        },
+        {"role": "user", "content": user_message}
+    ]
+
+    for i in range(max_iterations):
+        response = client.chat.completions.create(
+            model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
+            messages=messages,
+            tools=tools,
+            tool_choice="auto",
+            temperature=0.2
+        )
+
+        assistant_message = response.choices[0].message
+
+        if not assistant_message.tool_calls:
+            return assistant_message.content
+
+        messages.append(assistant_message)
+
+        for tool_call in assistant_message.tool_calls:
+            function_name = tool_call.function.name
+            arguments = json.loads(tool_call.function.arguments)
+
+            print(f"  Tool call: {function_name}({arguments})")
+
+            result = tool_functions[function_name](**arguments)
+
+            messages.append({
+                "role": "tool",
+                "tool_call_id": tool_call.id,
+                "content": json.dumps(result)
+            })
+
+    return "Max iterations reached"
+
+if __name__ == "__main__":
+    question = "What is the price of 50 bags of Yirgacheffe?"
+
+    print("User:", question)
+    print("\nProcessing...")
+    response = run_agent(question, mcp)
+    print("\nAssistant:", response)
+
+# Single-tool query
+run_agent("Tell me about the Geisha Reserve", mcp)
+
+# Different bulk quantity
+run_agent("How much for 100 bags of house blend?", mcp)

mcp-tools/solution-files/product_server.py

@@ -0,0 +1,75 @@
+from fastmcp import FastMCP
+from decimal import Decimal
+
+# Initialize the MCP server
+mcp = FastMCP("GlobalJava Product Tools")
+
+PRODUCTS = {
+    "ethiopian-yirgacheffe": {
+        "name": "Ethiopian Yirgacheffe Single-Origin",
+        "origin": "Yirgacheffe region, Ethiopia",
+        "flavor_profile": "Bright citrus, floral aroma, light body",
+        "price": 18.99,
+        "certifications": ["Fair Trade", "Organic"]
+    },
+    "house-blend": {
+        "name": "GlobalJava House Blend",
+        "origin": "Colombia and Brazil blend",
+        "flavor_profile": "Balanced, chocolatey, nutty",
+        "price": 12.99,
+        "certifications": []
+    },
+    "geisha-reserve": {
+        "name": "Limited Edition Geisha Reserve",
+        "origin": "Hacienda La Esmeralda, Panama",
+        "flavor_profile": "Jasmine, bergamot, white peach",
+        "price": 89.99,
+        "certifications": ["Single Estate", "Competition Grade"]
+    }
+}
+
+@mcp.tool()
+def get_product_info(product_id: str) -> dict:
+    """Get detailed information about a GlobalJava Roasters product including
+    name, origin, flavor profile, and current price. Use this when a customer
+    asks about a specific product.
+
+    Args:
+        product_id: The product identifier, e.g., 'ethiopian-yirgacheffe', 'house-blend'
+    """
+    if product_id not in PRODUCTS:
+        return {"error": f"Product '{product_id}' not found"}
+    return PRODUCTS[product_id]
+
+@mcp.tool()
+def calculate_bulk_price(product_id: str, quantity: int) -> dict:
+    """Calculate total price for a bulk order with volume discounts.
+    Discounts: 5% for 25+ bags, 10% for 50+ bags, 15% for 100+ bags.
+
+    Args:
+        product_id: The product identifier
+        quantity: Number of bags to order
+    """
+    if product_id not in PRODUCTS:
+        return {"error": f"Product '{product_id}' not found"}
+
+    base_price = Decimal(str(PRODUCTS[product_id]["price"]))
+
+    if quantity >= 100:
+        discount = Decimal("0.15")
+    elif quantity >= 50:
+        discount = Decimal("0.10")
+    elif quantity >= 25:
+        discount = Decimal("0.05")
+    else:
+        discount = Decimal("0")
+
+    unit_price = base_price * (1 - discount)
+    total_price = unit_price * quantity
+
+    return {
+        "quantity": quantity,
+        "discount_percent": float(discount * 100),
+        "unit_price": float(unit_price),
+        "total_price": float(total_price)
+    }

mcp-tools/starter-code/product_assistant_with_tools.py

@@ -0,0 +1,122 @@
+from openai import OpenAI
+import json
+import os
+from products import get_product_info, calculate_bulk_price
+
+TOOL_FUNCTIONS = {
+    "get_product_info": get_product_info,
+    "calculate_bulk_price": calculate_bulk_price
+}
+
+def execute_tool(function_name: str, arguments: dict) -> str:
+    """Safely execute a tool and return JSON result."""
+    if function_name not in TOOL_FUNCTIONS:
+        return json.dumps({"error": f"Unknown function: {function_name}"})
+
+    try:
+        result = TOOL_FUNCTIONS[function_name](**arguments)
+        return json.dumps(result)
+    except TypeError as e:
+        return json.dumps({"error": f"Invalid arguments: {e}"})
+    except Exception as e:
+        return json.dumps({"error": f"Tool execution failed: {e}"})
+
+client = OpenAI(
+    api_key=os.environ.get("TOGETHER_API_KEY"),
+    base_url="https://api.together.xyz/v1"
+)
+
+tools = [
+    {
+        "type": "function",
+        "function": {
+            "name": "get_product_info",
+            "description": "Get detailed information about a GlobalJava Roasters product including name, origin, flavor profile, and current price. Use this when a customer asks about a specific product.",
+            "strict": True,
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "product_id": {
+                        "type": "string",
+                        "description": "The product identifier, e.g., 'ethiopian-yirgacheffe', 'house-blend', 'geisha-reserve'"
+                    }
+                },
+                "required": ["product_id"],
+                "additionalProperties": False
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "calculate_bulk_price",
+            "description": "Calculate total price for a bulk order with volume discounts. Discounts: 5% for 25+ bags, 10% for 50+ bags, 15% for 100+ bags.",
+            "strict": True,
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "product_id": {
+                        "type": "string",
+                        "description": "The product identifier"
+                    },
+                    "quantity": {
+                        "type": "integer",
+                        "description": "Number of bags to order"
+                    }
+                },
+                "required": ["product_id", "quantity"],
+                "additionalProperties": False
+            }
+        }
+    }
+]
+
+def run_agent(user_message: str, tools: list, max_iterations: int = 10) -> str:
+    """Run the agentic loop until the model produces a final response."""
+
+    messages = [
+        {"role": "system", "content": "You are a product assistant for GlobalJava Roasters. Always use the available tools to look up current product information and pricing. Do not rely on general knowledge about coffee."},
+        {"role": "user", "content": user_message}
+    ]
+
+    for i in range(max_iterations):
+        response = client.chat.completions.create(
+            model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
+            messages=messages,
+            tools=tools,
+            tool_choice="auto",
+            temperature=0.2
+        )
+
+        assistant_message = response.choices[0].message
+
+        # No tool calls means we're done
+        if not assistant_message.tool_calls:
+            return assistant_message.content
+
+        messages.append(assistant_message)
+
+        # Process each tool call
+        for tool_call in assistant_message.tool_calls:
+            function_name = tool_call.function.name
+            arguments = json.loads(tool_call.function.arguments)
+
+            print(f"  Tool call: {function_name}({arguments})")
+
+            result = execute_tool(function_name, arguments)
+
+            messages.append({
+                "role": "tool",
+                "tool_call_id": tool_call.id,
+                "content": result  # Already JSON-serialized by execute_tool
+            })
+
+    return "Max iterations reached"
+
+if __name__ == "__main__":
+    question = "What is the price of 50 bags of Yirgacheffe?"
+
+    print("User:", question)
+    print("\nProcessing...")
+    response = run_agent(question, tools)
+    print("\nAssistant:", response)

mcp-tools/starter-code/products.py

@@ -0,0 +1,59 @@
+# products.py
+PRODUCTS = {
+    "ethiopian-yirgacheffe": {
+        "name": "Ethiopian Yirgacheffe Single-Origin",
+        "origin": "Yirgacheffe region, Ethiopia",
+        "flavor_profile": "Bright citrus, floral aroma, light body",
+        "price": 18.99,
+        "certifications": ["Fair Trade", "Organic"]
+    },
+    "house-blend": {
+        "name": "GlobalJava House Blend",
+        "origin": "Colombia and Brazil blend",
+        "flavor_profile": "Balanced, chocolatey, nutty",
+        "price": 12.99,
+        "certifications": []
+    },
+    "geisha-reserve": {
+        "name": "Limited Edition Geisha Reserve",
+        "origin": "Hacienda La Esmeralda, Panama",
+        "flavor_profile": "Jasmine, bergamot, white peach",
+        "price": 89.99,
+        "certifications": ["Single Estate", "Competition Grade"]
+    }
+}
+
+def get_product_info(product_id: str) -> dict:
+    """Look up product information by ID."""
+    if product_id not in PRODUCTS:
+        return {"error": f"Product '{product_id}' not found"}
+    return PRODUCTS[product_id]
+
+from decimal import Decimal
+
+def calculate_bulk_price(product_id: str, quantity: int) -> dict:
+    """Calculate price with volume discounts."""
+    if product_id not in PRODUCTS:
+        return {"error": f"Product '{product_id}' not found"}
+
+    base_price = Decimal(str(PRODUCTS[product_id]["price"]))
+
+    # Apply volume discounts
+    if quantity >= 100:
+        discount = Decimal("0.15")
+    elif quantity >= 50:
+        discount = Decimal("0.10")
+    elif quantity >= 25:
+        discount = Decimal("0.05")
+    else:
+        discount = Decimal("0")
+
+    unit_price = base_price * (1 - discount)
+    total_price = unit_price * quantity
+
+    return {
+        "quantity": quantity,
+        "discount_percent": float(discount * 100),
+        "unit_price": float(unit_price),
+        "total_price": float(total_price)
+    }