selvin-paul-raj/AI-Agent-MCP-Server

---
config:
  layout: fixed
---
flowchart TD
    UI_Input["📝 User Input"] --> Auth["🔐 Authentication"]
    Auth --> Router["🧭 Intelligent Router"]
    Router --> Validator["✅ Health Validator"]
    Validator --> Context["🧠 Context Manager"]
    Context --> PromptEngine["🧩 Prompt Engine"]
    PromptEngine --> PluginLoader["🔌 Plugin Loader"]
    PluginLoader --> Sandbox["🧪 Sandbox Executor"]
    
    Sandbox --> SearchAgent["� Search Agent"]
    Sandbox --> SummaryAgent["📄 Summary Agent"]
    Sandbox --> ClassifyAgent["🏷️ Classifier Agent"]
    Sandbox --> ScraperAgent["🌐 Web Scraper"]
    Sandbox --> MatcherAgent["👥 Role Matcher"]
    Sandbox --> CustomAgent["⚙️ Custom Agents"]
    
    SearchAgent --> TokenTracker["💰 Token Tracker"]
    SummaryAgent --> TokenTracker
    ClassifyAgent --> TokenTracker
    ScraperAgent --> TokenTracker
    MatcherAgent --> TokenTracker
    CustomAgent --> TokenTracker
    
    TokenTracker --> FallbackManager["🛟 Fallback Manager"]
    FallbackManager --> Analytics["📊 Analytics Engine"]
    Analytics --> Output["📦 Structured Output"]
    
    subgraph "Monitoring & Observability"
        Metrics["📈 Metrics"]
        Logs["📝 Logs"]
        Alerts["🚨 Alerts"]
    end
    
    TokenTracker --> Metrics
    FallbackManager --> Logs
    Analytics --> Alerts

# Clone the repository
git clone https://github.com/selvin-paul-raj/AI-Agent-MCP-Server.git
cd AI-Agent-MCP-Server

# Install UV (if not already installed)
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create virtual environment and install dependencies
# Create virtual environment and install dependencies
uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
uv pip install -r requirements.txt

# Copy environment template
cp .env.example .env

# Configure your environment variables
# Edit .env file with your API keys and settings

# AI Provider Configuration
GEMINI_API_KEY=your_gemini_api_key_here
GEMINI_MODEL=gemini-1.5-pro-latest

# Server Configuration
HOST=0.0.0.0
PORT=8000
WORKERS=4
DEBUG=false

# Security
JWT_SECRET_KEY=your_secure_jwt_secret_here
JWT_ALGORITHM=HS256
JWT_EXPIRE_MINUTES=60

# Database & Caching (Optional)
REDIS_URL=redis://localhost:6379/0
DATABASE_URL=sqlite:///./mcp_server.db

# Monitoring & Observability
ENABLE_METRICS=true
LOG_LEVEL=INFO
SENTRY_DSN=your_sentry_dsn_here

# Cost Management
MAX_TOKENS_PER_REQUEST=4096
COST_BUDGET_USD=100.0
ENABLE_COST_ALERTS=true

# Start the production server
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

# Development server with hot reload
uvicorn main:app --reload --host 127.0.0.1 --port 8000

# Interactive CLI mode
python cli.py

# Direct agent execution
python cli.py run-agent --name "search" --input "Find latest AI research papers"

# Batch processing
python cli.py batch --config batch_config.yaml --output results.json

# Health check
python cli.py health-check --verbose

from mcp_server import MCPClient, AgentRequest

# Initialize client
client = MCPClient(base_url="http://localhost:8000", api_key="your_api_key")

# Single agent request
request = AgentRequest(
    query="Summarize the latest tech news",
    agent_preference="summarizer",
    context={"user_id": "user123", "session_id": "session456"}
)

response = await client.execute_agent(request)
print(f"Agent: {response.agent_used}")
print(f"Result: {response.output}")
print(f"Cost: ${response.cost_usd:.4f}")
print(f"Tokens: {response.tokens_used}")

# Streaming response
async for chunk in client.stream_agent(request):
    print(chunk.content, end="", flush=True)

# Health check
curl -X GET "http://localhost:8000/health" \
  -H "Authorization: Bearer your_jwt_token"

# Execute agent
curl -X POST "http://localhost:8000/v1/agents/execute" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer your_jwt_token" \
  -d '{
    "query": "Analyze this business proposal",
    "agent": "analyzer",
    "context": {"priority": "high"},
    "options": {
      "max_tokens": 2048,
      "temperature": 0.7,
      "enable_fallback": true
    }
  }'

# Get agent metrics
curl -X GET "http://localhost:8000/v1/agents/metrics" \
  -H "Authorization: Bearer your_jwt_token"

from typing import Dict, Any
from core.base_agent import BaseAgent
from schemas.types import AgentResponse, AgentConfig

class MyCustomAgent(BaseAgent):
    """Custom agent for specialized tasks."""
    
    def __init__(self, config: AgentConfig):
        super().__init__(config)
        self.name = "my_custom_agent"
        self.description = "Handles custom business logic"
        self.version = "1.0.0"
    
    async def execute(self, input_data: Dict[str, Any]) -> AgentResponse:
        """Main execution logic."""
        try:
            # Your custom logic here
            query = input_data.get("query", "")
            context = input_data.get("context", {})
            
            # Process the request
            result = await self._process_query(query, context)
            
            return AgentResponse(
                success=True,
                output=result,
                metadata={
                    "agent": self.name,
                    "execution_time": self.execution_time,
                    "tokens_used": self.tokens_used
                }
            )
        except Exception as e:
            return AgentResponse(
                success=False,
                error=str(e),
                metadata={"agent": self.name}
            )
    
    async def _process_query(self, query: str, context: Dict) -> str:
        """Custom processing logic."""
        # Implement your business logic
        return f"Processed: {query}"
    
    def get_capabilities(self) -> Dict[str, Any]:
        """Return agent capabilities."""
        return {
            "input_types": ["text", "json"],
            "output_types": ["text", "structured"],
            "max_tokens": 4096,
            "languages": ["en", "es", "fr"],
            "specializations": ["custom_task", "business_logic"]
        }

You are a specialized AI assistant for {{ task_type }}.

Context:
{% if context %}
{% for key, value in context.items() %}
- {{ key }}: {{ value }}
{% endfor %}
{% endif %}

User Query: {{ query }}

Instructions:
- Be precise and accurate
- Use the provided context
- Format response as {{ output_format }}
{% if special_instructions %}
- {{ special_instructions }}
{% endif %}

Response:

agents:
  my_custom_agent:
    class_path: "agents.my_custom_agent.MyCustomAgent"
    enabled: true
    priority: 5
    timeout: 30
    retry_count: 3
    fallback_agents: ["backup_agent"]
    cost_limits:
      max_tokens: 4096
      max_cost_usd: 0.50
    config:
      model: "gemini-1.5-pro"
      temperature: 0.7
      max_tokens: 2048
    capabilities:
      - "custom_processing"
      - "business_logic"
    routing_keywords:
      - "custom"
      - "business"
      - "specialized"

# Unit tests
python -m pytest tests/agents/test_my_custom_agent.py -v

# Integration test
python cli.py test-agent --name my_custom_agent --input "test query"

# Load test
python cli.py benchmark --agent my_custom_agent --requests 100

# Prometheus metrics endpoint
GET /metrics

# Sample metrics
mcp_requests_total{agent="search",status="success"} 1,234
mcp_request_duration_seconds_bucket{agent="search",le="0.5"} 856
mcp_tokens_used_total{agent="summarizer"} 45,678
mcp_cost_usd_total{agent="classifier"} 12.34

# config/alerts.yaml
alerts:
  high_error_rate:
    condition: "error_rate > 0.05"
    duration: "5m"
    severity: "warning"
    
  cost_budget_exceeded:
    condition: "daily_cost_usd > 100"
    duration: "1m"
    severity: "critical"
    
  agent_down:
    condition: "agent_health_score < 0.8"
    duration: "2m"
    severity: "warning"

# JWT-based authentication
from auth.jwt_handler import JWTHandler

# Role-based access control
roles = {
    "admin": ["read", "write", "delete", "manage"],
    "developer": ["read", "write", "test"],
    "analyst": ["read", "execute"],
    "viewer": ["read"]
}

# API key management
headers = {
    "Authorization": "Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...",
    "X-API-Version": "v1",
    "X-Request-ID": "req_123456789"
}

# Agent execution limits
sandbox_config = {
    "max_execution_time": 30,  # seconds
    "max_memory_mb": 512,
    "max_cpu_percent": 50,
    "network_access": False,
    "file_system_access": "readonly",
    "allowed_imports": ["pandas", "numpy", "requests"]
}

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY . .

RUN pip install uv && uv pip install -r requirements.txt
EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# Build and run
docker build -t mcp-server .
docker run -p 8000:8000 --env-file .env mcp-server

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: mcp-server
  template:
    metadata:
      labels:
        app: mcp-server
    spec:
      containers:
      - name: mcp-server
        image: mcp-server:latest
        ports:
        - containerPort: 8000
        env:
        - name: GEMINI_API_KEY
          valueFrom:
            secretKeyRef:
              name: mcp-secrets
              key: gemini-api-key
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "1Gi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5

# k8s/hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: mcp-server-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: mcp-server
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

# Unit tests with coverage
pytest tests/unit/ --cov=mcp_server --cov-report=html

# Integration tests
pytest tests/integration/ -v

# Load testing
locust -f tests/load/locustfile.py --host=http://localhost:8000

# Security scan
bandit -r mcp_server/
safety check

# tests/config.yaml
test_environments:
  unit:
    database_url: "sqlite:///:memory:"
    redis_url: "redis://localhost:6379/15"
    
  integration:
    database_url: "sqlite:///test.db"
    mock_ai_provider: true
    
  load:
    target_rps: 100
    duration: "10m"
    concurrent_users: 50

{
  "daily_usage": {
    "date": "2025-08-06",
    "total_requests": 15420,
    "total_tokens": 2456789,
    "total_cost_usd": 45.67,
    "agents": {
      "search": {
        "requests": 6780,
        "tokens": 1234567,
        "cost_usd": 23.45,
        "avg_tokens_per_request": 182
      },
      "summarizer": {
        "requests": 4560,
        "tokens": 789012,
        "cost_usd": 15.67,
        "avg_tokens_per_request": 173
      }
    }
  },
  "cost_optimization": {
    "potential_savings": 12.34,
    "recommendations": [
      "Consider using gemini-1.5-flash for simple queries",
      "Implement prompt caching for repeated patterns",
      "Use agent fallback to cheaper models when appropriate"
    ]
  }
}

# Cost controls
cost_config = {
    "daily_budget_usd": 100.0,
    "per_user_budget_usd": 10.0,
    "alert_thresholds": {
        "warning": 0.8,  # 80% of budget
        "critical": 0.95  # 95% of budget
    },
    "auto_throttle": True,
    "fallback_to_cache": True
}

# Fork and clone the repository
git clone https://github.com/your-username/AI-Agent-MCP-Server.git
cd AI-Agent-MCP-Server

# Install development dependencies
uv pip install -r requirements-dev.txt

# Install pre-commit hooks
pre-commit install

# Run development server
uvicorn main:app --reload --port 8000

# Template for new agents
class NewAgent(BaseAgent):
    """
    Agent description and purpose.
    
    Capabilities:
    - Feature 1
    - Feature 2
    
    Input Format:
    - Expected input structure
    
    Output Format:  
    - Expected output structure
    """
    
    def __init__(self, config: AgentConfig):
        super().__init__(config)
        # Initialize agent-specific components
    
    async def execute(self, input_data: Dict[str, Any]) -> AgentResponse:
        # Implementation
        pass
    
    def get_capabilities(self) -> Dict[str, Any]:
        # Return capabilities
        pass

# Official Python SDK
pip install mcp-server-sdk

from mcp_sdk import MCPClient

client = MCPClient("http://localhost:8000", api_key="your_key")
response = await client.agents.execute(query="Hello", agent="chat")