version: '3'
tasks:
  default:
    desc: "Default task - runs all tasks"
    cmds:
      - task: all

  setup:
    desc: "Set up the Python environment"
    cmds:
      - pyenv install -s 3.12.9
      - pyenv local 3.12.9
      - poetry install
      - poetry config virtualenvs.in-project true
      - chmod +x .venv/bin/activate
      - source .venv/bin/activate

  run:
    desc: "Run the main example"
    cmds:
      - poetry run python src/main.py
    env:
      LLM_PROVIDER: ollama
      OLLAMA_MODEL: gemma3:27b

poetry init --name mcp-customer-service --version 0.1.0
poetry add fastmcp pydantic
poetry add --group dev pytest black isort

import asyncio
import logging
from datetime import datetime
from typing import List, Optional

from fastmcp import FastMCP
from pydantic import BaseModel, field_validator
from pydantic_core import PydanticCustomError


# Configure logging for better debugging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# Initialize FastMCP server
mcp = FastMCP("Customer Service Assistant")


# Data models for type safety and validation
class Customer(BaseModel):
    id: str
    name: str
    email: str
    phone: Optional[str] = None
    account_status: str = "active"
    last_interaction: Optional[datetime] = None

    @field_validator("email", mode="after")  # noqa
    @classmethod
    def email_must_be_valid(cls, v: str) -> str:
        if "@" not in v:
            raise PydanticCustomError(
                "invalid_email",
                "Invalid email format: {email} must contain @",
                {"email": v},
            )
        return v

class TicketRequest(BaseModel):
    customer_id: str
    subject: str
    description: str
    priority: str = "normal"

    @field_validator("priority", mode="after")  # noqa
    @classmethod
    def priority_must_be_valid(cls, v: str) -> str:
        valid_priorities = ["low", "normal", "high", "urgent"]
        if v not in valid_priorities:
            raise PydanticCustomError(
                "invalid_priority",
                "Priority must be one of: {valid_priorities}, got {priority}",
                {"valid_priorities": ", ".join(valid_priorities), "priority": v},
            )
        return v


# Simulated customer database
CUSTOMERS_DB = {
    "12345": Customer(
        id="12345",
        name="Alice Johnson",
        email="alice@example.com",
        phone="+1-555-0123",
        account_status="active",
        last_interaction=datetime.now(),
    ),
    "67890": Customer(
        id="67890",
        name="Bob Smith",
        email="bob@example.com",
        account_status="suspended",
    ),
}


# MCP Resource: Customer Data Access
@mcp.resource("customer://{customer_id}")
async def get_customer_info(customer_id: str) -> Customer:
    """Retrieve customer information by ID."""
    logger.info(f"Retrieving customer info for ID: {customer_id}")

    if customer_id not in CUSTOMERS_DB:
        raise ValueError(f"Customer {customer_id} not found")

    # Simulate database delay
    await asyncio.sleep(0.1)
    return CUSTOMERS_DB[customer_id]

@mcp.tool()
async def get_recent_customers(limit: int = 10) -> List[Customer]:
    """Retrieve recently active customers."""
    logger.info(f"Retrieving {limit} recent customers")

    # Sort by last interaction, return most recent
    sorted_customers = sorted(
        CUSTOMERS_DB.values(),
        key=lambda c: c.last_interaction or datetime.min,
        reverse=True,
    )

    return sorted_customers[:limit]


# MCP Tool: Create Support Ticket
@mcp.tool()
async def create_support_ticket(request: TicketRequest) -> dict:
    """Create a new customer support ticket."""
    logger.info(f"Creating ticket for customer {request.customer_id}")

    # Validate customer exists
    if request.customer_id not in CUSTOMERS_DB:
        raise ValueError(f"Customer {request.customer_id} not found")

    # Simulate ticket creation
    ticket_id = f"TICKET-{datetime.now().strftime('%Y%m%d%H%M%S')}"

    ticket = {
        "ticket_id": ticket_id,
        "customer_id": request.customer_id,
        "subject": request.subject,
        "description": request.description,
        "priority": request.priority,
        "status": "open",
        "created_at": datetime.now().isoformat(),
    }

    return ticket


# MCP Tool: Calculate Account Value
@mcp.tool()
async def calculate_account_value(
    customer_id: str, purchase_history: List[float]
) -> dict:
    """Calculate total account value and average purchase."""
    logger.info(f"Calculating account value for {customer_id}")

    if not purchase_history:
        return {
            "customer_id": customer_id,
            "total_value": 0.0,
            "average_purchase": 0.0,
            "purchase_count": 0,
        }

    total = sum(purchase_history)
    average = total / len(purchase_history)

    return {
        "customer_id": customer_id,
        "total_value": round(total, 2),
        "average_purchase": round(average, 2),
        "purchase_count": len(purchase_history),
    }


# MCP Prompt: Customer Service Response Template
@mcp.prompt("customer_service_response")
async def generate_service_response_prompt(
    customer_name: str, issue_type: str, resolution_steps: List[str]
) -> str:
    """Generate a professional customer service response."""

    steps_text = "\n".join(
        [f"{i+1}. {step}" for i, step in enumerate(resolution_steps)]
    )

    return f"""
You are a professional customer service representative.
Generate a helpful and empathetic response for the customer.

Customer: {customer_name}
Issue Type: {issue_type}

Resolution Steps:
{steps_text}

Guidelines:
- Be professional but warm
- Acknowledge the customer's concern
- Provide clear, actionable steps
- End with an offer for further assistance
- Keep the tone positive and solution-focused

Generate a complete customer service response
following these guidelines.
"""

def main():
    """Main entry point for the MCP server."""
    print("🚀 Starting Customer Service MCP Server...")
    print("📋 Available Resources:")
    print("   - customer://{customer_id} - Get customer info")
    print("🔧 Available Tools:")
    print("   - get_recent_customers - Get recent customers")
    print("   - create_support_ticket - Create support ticket")
    print("   - calculate_account_value - Calculate account value")
    print("📝 Available Prompts:")
    print("   - customer_service_response - Generate responses")
    print("\n✅ Server ready for connections!")

    # Run the server
    mcp.run()

if __name__ == "__main__":
    main()

import asyncio
import logging
from typing import List, Optional
from datetime import datetime
from fastmcp import FastMCP
from pydantic import BaseModel, validator


# Configure logging for better debugging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# Initialize FastMCP server
mcp = FastMCP("Customer Service Assistant")

class Customer(BaseModel):
    id: str
    name: str
    email: str
    phone: Optional[str] = None
    account_status: str = "active"
    last_interaction: Optional[datetime] = None

    @validator('email')
    def email_must_be_valid(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email format')
        return v

```**Pydantic models**define the structure of our data and provide automatic validation. Pydantic supports**JSON Schema**generation and**JSON serialization**, making it perfect for API development. The `@validator` decorator adds custom validation rules—here ensuring emails contain an '@' symbol. The `Optional` type hint indicates fields that can be `None`.


### 3. Simulated Database

```python
CUSTOMERS_DB = {
    "12345": Customer(
        id="12345",
        name="Alice Johnson",
        email="alice@example.com",
        phone="+1-555-0123",
        account_status="active",
        last_interaction=datetime.now()
    ),
    "67890": Customer(
        id="67890",
        name="Bob Smith",
        email="bob@example.com",
        account_status="suspended"
    )
}

@mcp.resource("customer://{customer_id}")
async def get_customer_info(customer_id: str) -> Customer:
    """Retrieve customer information by ID."""
    logger.info(f"Retrieving customer info for ID: {customer_id}")

    if customer_id not in CUSTOMERS_DB:
        raise ValueError(f"Customer {customer_id} not found")

    await asyncio.sleep(0.1)  # Simulate database delay
    return CUSTOMERS_DB[customer_id]

@mcp.tool()
async def create_support_ticket(request: TicketRequest) -> dict:
    """Create a new customer support ticket."""
    logger.info(f"Creating ticket for customer {request.customer_id}")

    if request.customer_id not in CUSTOMERS_DB:
        raise ValueError(f"Customer {request.customer_id} not found")

    ticket_id = f"TICKET-{datetime.now().strftime('%Y%m%d%H%M%S')}"
    # ... create ticket logic
    return ticket

@mcp.prompt("customer_service_response")
async def generate_service_response_prompt(
    customer_name: str,
    issue_type: str,
    resolution_steps: List[str]
) -> str:
    """Generate a professional customer service response."""

    steps_text = "\n".join([f"{i+1}. {step}"
                           for i, step in enumerate(resolution_steps)])

    return f"""
You are a professional customer service representative...
Customer: {customer_name}
Issue Type: {issue_type}
Resolution Steps:
{steps_text}
..."""

async def main():
    """Main entry point for the MCP server."""
    print("🚀 Starting Customer Service MCP Server...")
    # ... status messages
    await mcp.run()

if __name__ == "__main__":
    asyncio.run(main())

task setup  # Set up the environment
task run    # Start the server

{
  "mcpServers": {
    "customer-service": {
      "command": "/Users/richardhightower/src/mcp_article1/run-mcp-server.sh"
    }
  }
}

"""OpenAI Agents SDK integration with MCP server."""

import asyncio
from agents import Agent, Runner
from agents.mcp import MCPServerStdio

from config import Config

async def run_customer_service_scenarios():
    """Demonstrate OpenAI Agents + MCP integration."""
    print("🤖 Setting up OpenAI Agents + MCP integration...")

    # Create MCP server connection with proper 
    # async context manager
    mcp_server = MCPServerStdio(
        params={
            "command": "poetry",
            "args": ["run", "python", "src/main.py"]
        },
        cache_tools_list=True,
        name="Customer Service Server",
        client_session_timeout_seconds=30  # Increase timeout for startup
    )

    # Use the MCP server within an async context manager
    async with mcp_server as server:
        # Create agent with the connected MCP server
        agent = Agent(
            name="Customer Service Agent",
            instructions="""You are a helpful customer service 
            assistant.
            Use the available tools to help customers with 
            their requests.
            Always be professional and empathetic.
            
            Available tools:
            - get_recent_customers: Get a list of recent 
              customers
            - create_support_ticket: Create support tickets 
              for customers
            - calculate_account_value: Calculate customer 
              account values
            
            When helping customers:
            1. Look up their information first when possible
            2. Create tickets for issues that need follow-up
            3. Calculate account values when discussing 
               billing or purchases
            4. Always provide clear, helpful responses""",
            mcp_servers=[server]
        )

        # Example customer service scenarios
        scenarios = [
            "Get a list of recent customers and summarize "
            "their status",
            "Create a high-priority support ticket for " 
            "customer 67890 about billing issues",
            "Calculate the account value for customer 12345 "
            "with purchases: $150, $300, $89"
        ]

        for i, scenario in enumerate(scenarios, 1):
            print(f"\n📞 Scenario {i}: {scenario}")
            try:
                # Run the agent with the scenario
                result = await Runner.run(agent, scenario)
                print(f"🤖 Agent Response: {result.final_output}")
            except Exception as e:
                print(f"❌ Error: {e}")

            print("-" * 60)

async def main():
    """Main entry point."""
    Config.validate()
    if Config.LLM_PROVIDER != "openai":
        print("OpenAI Agents example requires OpenAI."
        " Set LLM_PROVIDER=openai in .env")
        return

    await run_customer_service_scenarios()

if __name__ == "__main__":
    asyncio.run(main())

"""OpenAI integration with MCP server."""

import asyncio
import json
from contextlib import AsyncExitStack

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

from config import Config

class OpenAIMCPChatBot:
    def __init__(self, api_key: str):
        self.client = AsyncOpenAI(api_key=api_key)
        self.sessions = []
        self.exit_stack = AsyncExitStack()
        self.available_tools = []
        self.tool_to_session = {}

    async def connect_to_server(self, server_name: str, server_config: dict) :
        """Connect to a single MCP server."""
        try:
            server_params = StdioServerParameters(**server_config)
            stdio_transport = await self.exit_stack.enter_async_context(
                stdio_client(server_params)
            )
            read, write = stdio_transport
            session = await self.exit_stack.enter_async_context(
                ClientSession(read, write)
            )
            await session.initialize()
            self.sessions.append(session)

            # List available tools for this session
            response = await session.list_tools()
            tools = response.tools
            print(f"Connected to {server_name} with tools:", 
                                             [t.name for t in tools])

            for tool in tools:
                self.tool_to_session[tool.name] = session
                # Convert MCP tool to OpenAI tool format
                openai_tool = {
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.inputSchema,
                    },
                }
                self.available_tools.append(openai_tool)
        except Exception as e:
            print(f"Failed to connect to {server_name}: {e}")

    async def connect_to_servers(self):
        """Connect to all configured MCP servers."""
        try:
            with open("server_config.json", "r") as file:
                data = json.load(file)

            servers = data.get("mcpServers", {})
            for server_name, server_config in servers.items():
                await self.connect_to_server(server_name, server_config)
        except Exception as e:
            print(f"Error loading server configuration: {e}")
            raise

    async def process_query(self, query: str):
        """Process a query using OpenAI with MCP tools."""
        messages = [{"role": "user", "content": query}]

        response = await self.client.chat.completions.create(
            model=Config.OPENAI_MODEL,
            messages=messages,
            tools=self.available_tools if self.available_tools else None,
        )

        process_query = True
        while process_query:
            message = response.choices[0].message

            if message.content:
                print(message.content)

            # Handle tool calls
            if message.tool_calls:
                messages.append(
                    {
                        "role": "assistant",
                        "content": message.content,
                        "tool_calls": message.tool_calls,
                    }
                )

                for tool_call in message.tool_calls:
                    tool_name = tool_call.function.name
                    tool_args = json.loads(tool_call.function.arguments)

                    print(f"Calling tool {tool_name} with args {tool_args}")

                    # Use the correct session for this tool
                    session = self.tool_to_session[tool_name]
                    result = await session.call_tool(tool_name, 
                                                            arguments=tool_args)

                    messages.append(
                        {
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "content": str(result.content),
                        }
                    )

                # Get the next response
                response = await self.client.chat.completions.create(
                    model=Config.OPENAI_MODEL,
                    messages=messages,
                    tools=self.available_tools if self.available_tools else None,
                )
            else:
                process_query = False

    async def chat_loop(self):
        """Run an interactive chat loop"""
        print("\nOpenAI MCP Chatbot Started!")
        print("Type your queries or 'quit' to exit.")

        while True:
            try:
                query = input("\nQuery: ").strip()
                if query.lower() == "quit":
                    break
                await self.process_query(query)
            except Exception as e:
                print(f"\nError: {str(e)}")

    async def cleanup(self):
        """Cleanly close all resources."""
        await self.exit_stack.aclose()


# Usage example
async def main():
    Config.LLM_PROVIDER = "openai"
    Config.validate()
    chatbot = OpenAIMCPChatBot(api_key=Config.OPENAI_API_KEY)
    try:
        await chatbot.connect_to_servers()
        await chatbot.chat_loop()
    finally:
        await chatbot.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

class OpenAIMCPChatBot:
    def __init__(self, api_key: str):
        self.client = AsyncOpenAI(api_key=api_key)
        self.sessions = []
        self.exit_stack = AsyncExitStack()
        self.available_tools = []
        self.tool_to_session = {}

async def connect_to_server(self, server_name: str, server_config: dict):
    server_params = StdioServerParameters(**server_config)
    stdio_transport = await self.exit_stack.enter_async_context(
        stdio_client(server_params)
    )
    read, write = stdio_transport
    session = await self.exit_stack.enter_async_context(
        ClientSession(read, write)
    )
    await session.initialize()

 # List available tools for this session
    response = await session.list_tools()
    tools = response.tools
    
    for tool in tools:
        self.tool_to_session[tool.name] = session
        # Convert MCP tool to OpenAI tool format
        openai_tool = {
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema,
            },
        }
        self.available_tools.append(openai_tool)

 async def process_query(self, query: str):
        messages = [{"role": "user", "content": query}]
        response = await self.client.chat.completions.create(
            model=Config.OPENAI_MODEL,
            messages=messages,
            tools=self.available_tools if self.available_tools else None,
        )
        
        while process_query:
            message = response.choices[0].message
            if message.tool_calls:
                # Handle tool execution here
                for tool_call in message.tool_calls:
                    tool_name = tool_call.function.name
                    tool_args = json.loads(tool_call.function.arguments)
                    session = self.tool_to_session[tool_name]
                    result = await session.call_tool(tool_name, arguments=tool_args)

 async def chat_loop(self):
        while True:
            try:
                query = input("\nQuery: ").strip()
                if query.lower() == "quit":
                    break
                await self.process_query(query)
            except Exception as e:
                print(f"\nError: {str(e)}")

    async def cleanup(self):
        await self.exit_stack.aclose()

"""Anthropic integration with MCP server."""

import asyncio
import json
from contextlib import AsyncExitStack

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

from config import Config

class AnthropicMCPChatBot:
    def __init__(self, api_key: str):
        self.anthropic = Anthropic(api_key=api_key)
        self.sessions = []
        self.exit_stack = AsyncExitStack()
        self.available_tools = []
        self.tool_to_session = {}

    async def connect_to_server(self, server_name: str, server_config: dict) :
        """Connect to a single MCP server."""
        try:
            server_params = StdioServerParameters(**server_config)
            stdio_transport = await self.exit_stack.enter_async_context(
                stdio_client(server_params)
            )
            read, write = stdio_transport
            session = await self.exit_stack.enter_async_context(
                ClientSession(read, write)
            )
            await session.initialize()
            self.sessions.append(session)

            # List available tools for this session
            response = await session.list_tools()
            tools = response.tools
            print(f"Connected to {server_name} with tools:", 
			            [t.name for t in tools])

            for tool in tools:
                self.tool_to_session[tool.name] = session
                self.available_tools.append(
                    {
                        "name": tool.name,
                        "description": tool.description,
                        "input_schema": tool.inputSchema,
                    }
                )
        except Exception as e:
            print(f"Failed to connect to {server_name}: {e}")

    async def connect_to_servers(self):
        """Connect to all configured MCP servers."""
        try:
            with open("server_config.json", "r") as file:
                data = json.load(file)

            servers = data.get("mcpServers", {})
            for server_name, server_config in servers.items():
                await self.connect_to_server(server_name, server_config)
        except Exception as e:
            print(f"Error loading server configuration: {e}")
            raise

    async def process_query(self, query: str):
        """Process a query using Claude with MCP tools."""
        messages = [{"role": "user", "content": query}]

        response = self.anthropic.messages.create(
            max_tokens=2024,
            model=Config.ANTHROPIC_MODEL,
            tools=self.available_tools,
            messages=messages,
        )

        process_query = True
        while process_query:
            assistant_content = []
            for content in response.content:
                if content.type == "text":
                    print(content.text)
                    assistant_content.append(content)
                    if len(response.content) == 1:
                        process_query = False

                elif content.type == "tool_use":
                    assistant_content.append(content)
                    messages.append({"role": "assistant", 
                                     "content": assistant_content})

                    tool_id = content.id
                    tool_args = content.input
                    tool_name = content.name

                    print(f"Calling tool {tool_name} with args {tool_args}")

                    # Use the correct session for this tool
                    session = self.tool_to_session[tool_name]
                    result = await session.call_tool(tool_name,
                                                arguments=tool_args)

                    messages.append(
                        {
                            "role": "user",
                            "content": [
                                {
                                    "type": "tool_result",
                                    "tool_use_id": tool_id,
                                    "content": result.content,
                                }
                            ],
                        }
                    )

                    response = self.anthropic.messages.create(
                        max_tokens=2024,
                        model=Config.ANTHROPIC_MODEL,
                        tools=self.available_tools,
                        messages=messages,
                    )

                    if (
                        len(response.content) == 1
                        and response.content[0].type == "text"
                    ):
                        print(response.content[0].text)
                        process_query = False

    async def chat_loop(self):
        """Run an interactive chat loop"""
        print("\nAnthropic MCP Chatbot Started!")
        print("Type your queries or 'quit' to exit.")

        while True:
            try:
                query = input("\nQuery: ").strip()
                if query.lower() == "quit":
                    break
                await self.process_query(query)
            except Exception as e:
                print(f"\nError: {str(e)}")

    async def cleanup(self):
        """Cleanly close all resources."""
        await self.exit_stack.aclose()


# Usage example
async def main():
    Config.LLM_PROVIDER = "anthropic"
    Config.validate()

    chatbot = AnthropicMCPChatBot(api_key=Config.ANTHROPIC_API_KEY)
    try:
        await chatbot.connect_to_servers()
        await chatbot.chat_loop()
    finally:
        await chatbot.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

class AnthropicMCPChatBot:
    def __init__(self, api_key: str):
        self.anthropic = Anthropic(api_key=api_key)
        self.sessions = []
        self.exit_stack = AsyncExitStack()
        self.available_tools = []
        self.tool_to_session = {}

async def connect_to_server(self, server_name: str, server_config: dict):
    server_params = StdioServerParameters(**server_config)
    stdio_transport = await self.exit_stack.enter_async_context(
        stdio_client(server_params)
    )
    read, write = stdio_transport
    session = await self.exit_stack.enter_async_context(
        ClientSession(read, write)
    )
    await session.initialize()

 response = await session.list_tools()
    tools = response.tools
    print(f"Connected to {server_name} with tools:", [t.name for t in tools])

    for tool in tools:
        self.tool_to_session[tool.name] = session
        self.available_tools.append({
            "name": tool.name,
            "description": tool.description,
            "input_schema": tool.inputSchema,
        })

async def process_query(self, query: str):
    messages = [{"role": "user", "content": query}]
    response = self.anthropic.messages.create(
        max_tokens=2024,
        model=Config.ANTHROPIC_MODEL,
        tools=self.available_tools,
        messages=messages,
    )

 while process_query:
        assistant_content = []
        for content in response.content:
            if content.type == "tool_use":
                tool_id = content.id
                tool_args = content.input
                tool_name = content.name
                session = self.tool_to_session[tool_name]
                result = await session.call_tool(tool_name, arguments=tool_args)

{
  "mcpServers": {
    "customer-service": {
      "command": "poetry",
      "args": ["run", "python", "src/main.py"]
    }
  }
}

# Install dependencies
poetry add openai anthropic python-mcp-sdk python-dotenv


# Set up environment variables
echo "OPENAI_API_KEY=your-key-here" >> .env
echo "ANTHROPIC_API_KEY=your-key-here" >> .env


# Run OpenAI version
python openai_mcp_chatbot.py


# Run Anthropic version
python anthropic_mcp_chatbot.py

poetry add langchain langchain-mcp-adapters langchain-openai

"""LangChain integration with MCP server."""

import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

from config import Config

async def setup_langchain_mcp_agent():
    """Set up a LangChain agent with MCP tools."""

    # Initialize the language model
    llm = ChatOpenAI(
        model=Config.OPENAI_MODEL, temperature=0.1, api_key=Config.OPENAI_API_KEY
    )

    # Connect to our MCP server using MultiServerMCPClient
    client = MultiServerMCPClient(
        {
            "customer-service": {
                "command": "poetry",
                "args": ["run", "python", "src/main.py"],
                "transport": "stdio",
            }
        }
    )

    # Get all available tools from MCP servers
    tools = await client.get_tools()

    # Create a ReAct agent with the tools
    agent = create_react_agent(llm, tools)

    return agent, client

async def run_customer_service_scenarios():
    """Demonstrate LangChain + MCP integration."""
    print("🔗 Setting up LangChain + MCP integration...")

    agent, client = await setup_langchain_mcp_agent()

    # Example customer service scenarios
    scenarios = [
        "Look up customer 12345 and summarize their account status",
        "Create a high-priority support ticket for customer 67890 about billing",
        "Calculate account value for customer with purchases: $150, $300, $89",
    ]

    for scenario in scenarios:
        print(f"\n📞 Scenario: {scenario}")
        try:
            response = await agent.ainvoke(
                {"messages": [{"role": "user", "content": scenario}]}
            )

            # Extract the final AI response
            final_message = response["messages"][-1]
            if hasattr(final_message, "content"):
                print(f"🤖 Response: {final_message.content}")
            else:
                print(f"🤖 Response: {final_message}")

        except Exception as e:
            print(f"❌ Error: {e}")

        print("-" * 60)

async def main():
    """Main entry point."""
    Config.validate()

    await run_customer_service_scenarios()

if __name__ == "__main__":
    asyncio.run(main())

# Imports and configuration
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from config import Config

async def setup_langchain_mcp_agent():
    """Set up a LangChain agent with MCP tools."""
    # Initialize language model
    llm = ChatOpenAI(
        model=Config.OPENAI_MODEL, 
        temperature=0.1, 
        api_key=Config.OPENAI_API_KEY
    )
    return llm

 # Connect to MCP server
    client = MultiServerMCPClient({
        "customer-service": {
            "command": "poetry",
            "args": ["run", "python", "src/main.py"],
            "transport": "stdio",
        }
    })
    
    # Get available tools
    tools = await client.get_tools()
    
    # Create ReAct agent
    agent = create_react_agent(llm, tools)
    return agent, client

async def run_customer_service_scenarios():
    """Demonstrate LangChain + MCP integration."""
    print("🔗 Setting up LangChain + MCP integration...")
    agent, client = await setup_langchain_mcp_agent()
    
    scenarios = [
        "Look up customer 12345 and summarize their account status",
        "Create a high-priority support ticket for customer 67890",
        "Calculate account value for purchases: $150, $300, $89",
    ]

 for scenario in scenarios:
        print(f"\n📞 Scenario: {scenario}")
        try:
            response = await agent.ainvoke({
                "messages": [{"role": "user", "content": scenario}]
            })
            
            # Extract response
            final_message = response["messages"][-1]
            if hasattr(final_message, "content"):
                print(f"🤖 Response: {final_message.content}")
            else:
                print(f"🤖 Response: {final_message}")
        except Exception as e:
            print(f"❌ Error: {e}")
        print("-" * 60)

"""DSPy integration with MCP server."""

import asyncio

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

from config import Config


# Define a DSPy signature for our customer service tasks
class CustomerServiceSignature(dspy.Signature):
    """Handle customer service requests using available tools."""

    request: str = dspy.InputField(desc="Customer service request")
    response: str = dspy.OutputField(desc="Helpful customer service response")

async def setup_dspy_mcp_integration():
    """Set up DSPy with MCP tools."""

    # Configure DSPy with your preferred language model
    if Config.LLM_PROVIDER == "openai":
        llm = dspy.LM(f"openai/{Config.OPENAI_MODEL}", 
                                          api_key=Config.OPENAI_API_KEY)
    elif Config.LLM_PROVIDER == "anthropic":
        llm = dspy.LM(
            f"anthropic/{Config.ANTHROPIC_MODEL}", 
                                           api_key=Config.ANTHROPIC_API_KEY
        )
    else:
        print("DSPy requires OpenAI or Anthropic provider")
        return None

    dspy.configure(lm=llm)

    # Create MCP client connection
    server_params = StdioServerParameters(
        command="poetry", args=["run", "python", "src/main.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the MCP connection
            await session.initialize()

            # List available tools
            tools = await session.list_tools()

            # Convert MCP tools to DSPy tools
            dspy_tools = []
            for tool in tools.tools:
                dspy_tools.append(dspy.Tool.from_mcp_tool(session, tool))

            # Create a ReAct agent with the tools
            react = dspy.ReAct(CustomerServiceSignature, tools=dspy_tools)

            # Test the integration
            result = await react.acall(
                request="Look up customer 12345 and create a support ticket"
                " as the bbq grill that she bought is defective."
            )

            print(f"DSPy Result: {result}")

async def main():
    """Main entry point."""
    Config.validate()
    await setup_dspy_mcp_integration()

if __name__ == "__main__":
    asyncio.run(main())

"""DSPy integration with MCP server."""

import asyncio
import dspy
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from config import Config

# Define a DSPy signature for our customer service tasks
class CustomerServiceSignature(dspy.Signature):
    """Handle customer service requests using available tools."""
    
    request: str = dspy.InputField(desc="Customer service request")
    response: str = dspy.OutputField(desc="Helpful customer service response")

 # Configure DSPy with your preferred language model
    if Config.LLM_PROVIDER == "openai":
        llm = dspy.LM(f"openai/{Config.OPENAI_MODEL}", api_key=Config.OPENAI_API_KEY)
    elif Config.LLM_PROVIDER == "anthropic":
        llm = dspy.LM(
            f"anthropic/{Config.ANTHROPIC_MODEL}", api_key=Config.ANTHROPIC_API_KEY
        )
    else:
        print("DSPy requires OpenAI or Anthropic provider")
        return None

    dspy.configure(lm=llm)

 # Create MCP client connection
    server_params = StdioServerParameters(
        command="poetry", args=["run", "python", "src/main.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the MCP connection
            await session.initialize()

# List available tools
 tools = await session.list_tools()

 # Convert MCP tools to DSPy tools
 dspy_tools = []
 for tool in tools.tools:
   dspy_tools.append(dspy.Tool.from_mcp_tool(session, tool))

 # Create a ReAct agent with the tools
 react = dspy.ReAct(CustomerServiceSignature, tools=dspy_tools)

 # Test the integration
 result = await react.acall(
                request="Look up customer 12345 and create a support ticket"
                " as the bbq grill that she bought is defective.")

 print(f"DSPy Result: {result}")

 

async def main():
    """Main entry point."""
    Config.validate()
    await setup_dspy_mcp_integration()

if __name__ == "__main__":
    asyncio.run(main())

poetry add litellm

"""LiteLLM integration with MCP server."""

import asyncio
import json

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

from config import Config

async def setup_litellm_mcp():
    """Set up LiteLLM with MCP tools."""

    # Create MCP server connection
    server_params = StdioServerParameters(
        command="poetry", args=["run", "python", "src/main.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the MCP connection
            await session.initialize()

            # Load MCP tools in OpenAI format
            tools = await experimental_mcp_client.load_mcp_tools(
                session=session, format="openai"
            )

            print(f"Loaded {len(tools)} MCP tools")

            # Use tools with different models
            models_to_test = []

            if Config.LLM_PROVIDER == "openai":
                models_to_test.append(Config.OPENAI_MODEL)
            elif Config.LLM_PROVIDER == "anthropic":
                models_to_test.append(Config.ANTHROPIC_MODEL)
            else:
                models_to_test = [Config.OPENAI_MODEL, Config.ANTHROPIC_MODEL]

            for model in models_to_test:
                try:
                    print(f"\nTesting with {model}...")

                    # Initial conversation
                    messages = [
                        {
                            "role": "user",
                            "content": "Customer 67890 recently purchases "
                            "were $150, $300, $13 and $89. "
                            "Calculate their total account value.",
                        }
                    ]

                    # First call to get tool requests
                    response = await litellm.acompletion(
                        model=model,
                        messages=messages,
                        tools=tools,
                    )

                    # Extract the response
                    message = response.choices[0].message

                    # Check if the model made tool calls
                    if hasattr(message, "tool_calls") and message.tool_calls:
                        print(f"🔧 Tool calls made: {len(message.tool_calls)}")

                        # Add assistant's message with tool calls to conversation
                        messages.append(
                            {
                                "role": "assistant",
                                "content": message.content,
                                "tool_calls": message.tool_calls,
                            }
                        )

                        # Execute each tool call
                        for call in message.tool_calls:
                            print(f"   - Executing {call.function.name}")

                            # Execute the tool through MCP
                            arguments = json.loads(call.function.arguments)
                            result = await session.call_tool(
                                call.function.name, arguments
                            )

                            # Add tool result to conversation
                            messages.append(
                                {
                                    "role": "tool",
                                    "content": str(result.content),
                                    "tool_call_id": call.id,
                                }
                            )

                        # Get final response from model with tool results
                        final_response = await litellm.acompletion(
                            model=model,
                            messages=messages,
                            tools=tools,
                        )

                        final_content = final_response.choices[0].message.content
                        print(f"🤖 Final Response: {final_content}")

                    else:
                        # Display content if available (no tools called)
                        if message.content:
                            print(f"🤖 Response: {message.content}")
                        else:
                            print("🤖 Response: (No response)")

                except Exception as e:
                    print(f"Error with {model}: {e}")

async def main():
    """Main entry point."""
    Config.validate()
    await setup_litellm_mcp()

if __name__ == "__main__":
    asyncio.run(main())

"""LiteLLM integration with MCP server."""
import asyncio
import json
import litellm
from litellm import experimental_mcp_client
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from config import Config

async def setup_litellm_mcp():
    """Set up LiteLLM with MCP tools."""
    
    # Initialize server parameters
    server_params = StdioServerParameters(
        command="poetry",
        args=["run", "python", "src/main.py"]
    )

    # Create client session
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize MCP connection
            await session.initialize()
            
            # Load tools in OpenAI format
            tools = await experimental_mcp_client.load_mcp_tools(
                session=session,
                format="openai"
            )
            print(f"Loaded {len(tools)} MCP tools")

						# Select model based on provider config
            if Config.LLM_PROVIDER == "openai":
                models_to_test.append(Config.OPENAI_MODEL)
            elif Config.LLM_PROVIDER == "anthropic": 
                models_to_test.append(Config.ANTHROPIC_MODEL)
            else:
                models_to_test = [Config.OPENAI_MODEL, Config.ANTHROPIC_MODEL]

            # Test each configured model
            for model in models_to_test:
                try:
                    print(f"\nTesting with {model}...")
                    
                    # Set up initial test message
                    messages = [{
                        "role": "user",
                        "content": (
                            "Customer 67890 recently purchases "
                            "were $150, $300, $13 and $89. "
                            "Calculate their total account value."
                        )
                    }]

response = await litellm.acompletion(
                        model=model,
                        messages=messages,
                        tools=tools,
                    )
 message = response.choices[0].message

 if hasattr(message, "tool_calls") and message.tool_calls:
     print(f"🔧 Tool calls made: {len(message.tool_calls)}")
     messages.append(
                     {
                         "role": "assistant",
                         "content": message.content,
                         "tool_calls": message.tool_calls,
                     }
     )

for call in message.tool_calls:
	  print(f"   - Executing {call.function.name}")
	  arguments = json.loads(call.function.arguments)
	  result = await session.call_tool(call.function.name, arguments)
	  messages.append(
      {
          "role": "tool",
          "content": str(result.content),
          "tool_call_id": call.id,
      }
	  )
   ...                     
	 final_response = await litellm.acompletion(
                            model=model,
                            messages=messages,
                            tools=tools,
                        )

	 final_content = final_response.choices[0].message.content
	 print(f"🤖 Final Response: {final_content}")

async def main():
    """Main entry point."""
    Config.validate()
    await setup_litellm_mcp()

if __name__ == "__main__":
    asyncio.run(main())