LLM集成协议服务器

pip install -r requirements.txt

#!/usr/bin/env python3
"""
MCP-LLM Server: A Model Context Protocol server for LLM operations
"""

import asyncio
import json
import logging
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from abc import ABC, abstractmethod

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

@dataclass
class MCPMessage:
    """Represents an MCP protocol message"""
    id: str
    method: str
    params: Dict[str, Any]
    jsonrpc: str = "2.0"

@dataclass
class MCPResponse:
    """Represents an MCP protocol response"""
    id: str
    result: Any
    jsonrpc: str = "2.0"
    error: Optional[Dict[str, Any]] = None

class LLMProvider(ABC):
    """Abstract base class for LLM providers"""
    
    @abstractmethod
    async def generate(self, prompt: str, **kwargs) -> str:
        """Generate text using the LLM"""
        pass
    
    @abstractmethod
    async def chat(self, messages: List[Dict[str, str]], **kwargs) -> str:
        """Generate chat response"""
        pass

class MockLLMProvider(LLMProvider):
    """Mock LLM provider for testing"""
    
    async def generate(self, prompt: str, **kwargs) -> str:
        """Mock text generation"""
        await asyncio.sleep(0.1)  # Simulate API call
        return f"Generated response for: {prompt[:50]}..."
    
    async def chat(self, messages: List[Dict[str, str]], **kwargs) -> str:
        """Mock chat response"""
        await asyncio.sleep(0.1)  # Simulate API call
        last_message = messages[-1]["content"] if messages else "Hello"
        return f"Chat response to: {last_message[:50]}..."

class MCPLLMServer:
    """Main MCP server class for LLM operations"""
    
    def __init__(self, llm_provider: LLMProvider):
        self.llm_provider = llm_provider
        self.context_store: Dict[str, List[Dict[str, str]]] = {}
        self.tools = {
            "generate_text": self._generate_text,
            "chat_completion": self._chat_completion,
            "manage_context": self._manage_context,
            "list_contexts": self._list_contexts,
            "health_check": self._health_check
        }
        logger.info("MCP-LLM Server initialized")
    
    async def _generate_text(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Generate text using the LLM"""
        try:
            prompt = params.get("prompt", "")
            max_tokens = params.get("max_tokens", 100)
            temperature = params.get("temperature", 0.7)
            
            if not prompt:
                raise ValueError("Prompt is required")
            
            result = await self.llm_provider.generate(
                prompt=prompt,
                max_tokens=max_tokens,
                temperature=temperature
            )
            
            return {
                "text": result,
                "tokens_used": len(result.split()),
                "model": "mock-llm-v1"
            }
        except Exception as e:
            logger.error(f"Error in text generation: {e}")
            raise
    
    async def _chat_completion(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Handle chat completion requests"""
        try:
            messages = params.get("messages", [])
            context_id = params.get("context_id")
            
            if context_id and context_id in self.context_store:
                # Use stored context
                full_messages = self.context_store[context_id] + messages
            else:
                full_messages = messages
            
            result = await self.llm_provider.chat(
                messages=full_messages,
                **{k: v for k, v in params.items() if k not in ["messages", "context_id"]}
            )
            
            # Update context if context_id provided
            if context_id:
                if context_id not in self.context_store:
                    self.context_store[context_id] = []
                self.context_store[context_id].extend(messages)
                self.context_store[context_id].append({"role": "assistant", "content": result})
            
            return {
                "message": {"role": "assistant", "content": result},
                "context_id": context_id,
                "tokens_used": len(result.split())
            }
        except Exception as e:
            logger.error(f"Error in chat completion: {e}")
            raise
    
    async def _manage_context(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Manage conversation context"""
        action = params.get("action")
        context_id = params.get("context_id")
        
        if action == "create":
            context_id = context_id or f"ctx_{len(self.context_store)}"
            self.context_store[context_id] = []
            return {"context_id": context_id, "status": "created"}
        
        elif action == "clear" and context_id:
            if context_id in self.context_store:
                self.context_store[context_id] = []
                return {"context_id": context_id, "status": "cleared"}
            else:
                raise ValueError(f"Context {context_id} not found")
        
        elif action == "delete" and context_id:
            if context_id in self.context_store:
                del self.context_store[context_id]
                return {"context_id": context_id, "status": "deleted"}
            else:
                raise ValueError(f"Context {context_id} not found")
        
        else:
            raise ValueError("Invalid action or missing context_id")
    
    async def _list_contexts(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """List all available contexts"""
        contexts = []
        for ctx_id, messages in self.context_store.items():
            contexts.append({
                "id": ctx_id,
                "message_count": len(messages),
                "last_updated": "recent"  # In a real implementation, you'd track timestamps
            })
        return {"contexts": contexts}
    
    async def _health_check(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Health check endpoint"""
        return {
            "status": "healthy",
            "server": "MCP-LLM",
            "version": "1.0.0",
            "active_contexts": len(self.context_store)
        }
    
    async def handle_request(self, message: MCPMessage) -> MCPResponse:
        """Handle incoming MCP requests"""
        try:
            if message.method in self.tools:
                result = await self.tools[message.method](message.params)
                return MCPResponse(id=message.id, result=result)
            else:
                error = {
                    "code": -32601,
                    "message": f"Method not found: {message.method}",
                    "data": {"available_methods": list(self.tools.keys())}
                }
                return MCPResponse(id=message.id, result=None, error=error)
        
        except Exception as e:
            logger.error(f"Error handling request: {e}")
            error = {
                "code": -32603,
                "message": "Internal error",
                "data": {"error": str(e)}
            }
            return MCPResponse(id=message.id, result=None, error=error)
    
    async def start_server(self, host: str = "localhost", port: int = 8000):
        """Start the MCP server (placeholder for actual server implementation)"""
        logger.info(f"Starting MCP-LLM server on {host}:{port}")
        
        # In a real implementation, you would set up WebSocket or HTTP server here
        # For demonstration, we'll just show how requests would be handled
        
        # Example usage:
        sample_requests = [
            MCPMessage(id="1", method="health_check", params={}),
            MCPMessage(id="2", method="generate_text", params={"prompt": "Write a haiku about AI"}),
            MCPMessage(id="3", method="chat_completion", params={
                "messages": [{"role": "user", "content": "Hello, how are you?"}]
            })
        ]
        
        for request in sample_requests:
            response = await self.handle_request(request)
            logger.info(f"Request: {request.method}")
            logger.info(f"Response: {json.dumps(response.__dict__, indent=2)}")
            print("-" * 50)

# Usage Example
async def main():
    """Main function to demonstrate the MCP server"""
    # Initialize with mock LLM provider
    llm_provider = MockLLMProvider()
    server = MCPLLMServer(llm_provider)
    
    # Start the server (demo mode)
    await server.start_server()

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

# config.py
import os
from typing import Dict, Any

class Config:
    """Configuration settings for MCP-LLM server"""
    
    # Server settings
    HOST = os.getenv("MCP_HOST", "localhost")
    PORT = int(os.getenv("MCP_PORT", "8000"))
    
    # LLM Provider settings
    LLM_PROVIDER = os.getenv("LLM_PROVIDER", "mock")  # mock, openai, anthropic
    
    # OpenAI settings
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-3.5-turbo")
    
    # Rate limiting
    MAX_REQUESTS_PER_MINUTE = int(os.getenv("MAX_REQUESTS_PER_MINUTE", "60"))
    
    # Context management
    MAX_CONTEXT_LENGTH = int(os.getenv("MAX_CONTEXT_LENGTH", "4000"))
    MAX_CONTEXTS = int(os.getenv("MAX_CONTEXTS", "100"))
    
    @classmethod
    def to_dict(cls) -> Dict[str, Any]:
        """Convert config to dictionary"""
        return {
            key: getattr(cls, key)
            for key in dir(cls)
            if not key.startswith('_') and not callable(getattr(cls, key))
        }

{
  "jsonrpc": "2.0",
  "id": "1",
  "method": "generate_text",
  "params": {
    "prompt": "Explain quantum computing in simple terms",
    "max_tokens": 150,
    "temperature": 0.7
  }
}

python -m pytest tests/