Practical Implementation: Building Your First Integrated AI Ecosystem

 1
# Install Node.js and npm
 2
curl -fsSL https://deb.nodesource.com/setup_20.x | sudo -E bash -
 3
sudo apt install -y nodejs
 4
 5
# Install Python and pip
 6
sudo apt install -y python3 python3-pip python3-venv
 7
 8
# Install MCP development tools
 9
npm install -g @modelcontextprotocol/dev-tools
10
11
# Install Hermes Agent (for AI agent framework)
12
curl -fsSL https://raw.githubusercontent.com/NousResearch/hermes-agent/main/scripts/install.sh | bash

 1
content-studio/
 2
├── mcp-servers/
 3
│   ├── research-server/
 4
│   ├── writing-server/
 5
│   └── design-server/
 6
├── agents/
 7
│   ├── researcher/
 8
│   ├── writer/
 9
│   └── designer/
10
├── coordination/
11
│   └── orchestrator/
12
├── config/
13
│   └── ecosystem.yaml
14
└── README.md

 1
// mcp-servers/research-server/index.js
 2
const { Server } = require('@modelcontextprotocol/sdk/server');
 3
const { researchTools } = require('./tools/research');
 4
 5
const server = new Server(
 6
  { name: 'research-server', version: '1.0.0' },
 7
  { capabilities: { tools: {} } }
 8
);
 9
10
// Register research tools
11
server.setRequestHandler('tools/list', async () => ({
12
  tools: researchTools
13
}));
14
15
server.setRequestHandler('tools/call', async (request) => {
16
  const { name, arguments: args } = request.params;
17
  
18
  switch (name) {
19
    case 'search_web':
20
      return await searchWeb(args.query, args.limit);
21
    case 'analyze_trends':
22
      return await analyzeTrends(args.data);
23
    case 'summarize_content':
24
      return await summarizeContent(args.text);
25
    default:
26
      throw new Error(Tool ${name} not found);
27
  }
28
});
29
30
// Start server
31
server.listen(new StdioServerTransport()).catch(console.error);

 1
# mcp-servers/writing-server/server.py
 2
from mcp.server import Server, StdioServerTransport
 3
from mcp.server.models import Tool
 4
 5
server = Server("writing-server", "1.0.0")
 6
 7
writing_tools = [
 8
    Tool(
 9
        name="generate_article",
10
        description="Generate an article on a given topic",
11
        inputSchema={
12
            "type": "object",
13
            "properties": {
14
                "topic": {"type": "string"},
15
                "length": {"type": "string", "enum": ["short", "medium", "long"]},
16
                "tone": {"type": "string", "enum": ["formal", "casual", "persuasive"]}
17
            },
18
            "required": ["topic"]
19
        }
20
    ),
21
    Tool(
22
        name="improve_text",
23
        description="Improve existing text for clarity and engagement",
24
        inputSchema={
25
            "type": "object",
26
            "properties": {
27
                "text": {"type": "string"},
28
                "goal": {"type": "string", "enum": ["clarity", "engagement", "seo"]}
29
            },
30
            "required": ["text"]
31
        }
32
    )
33
]
34
35
@server.list_tools()
36
async def handle_list_tools():
37
    return writing_tools
38
39
@server.call_tool()
40
async def handle_tool_call(name: str, arguments: dict):
41
    if name == "generate_article":
42
        return await generate_article(arguments)
43
    elif name == "improve_text":
44
        return await improve_text(arguments)
45
    else:
46
        raise ValueError(f"Unknown tool: {name}")
47
48
async def main():
49
    async with server.run_transport(StdioServerTransport()):
50
        await server.wait_until_exit()
51
52
if __name__ == "__main__":
53
    asyncio.run(main())

 1
// mcp-servers/design-server/index.js
 2
const { createCanvas } = require('canvas');
 3
const { Server } = require('@modelcontextprotocol/sdk/server');
 4
 5
const designTools = [
 6
  {
 7
    name: 'create_banner',
 8
    description: 'Create a banner image for social media',
 9
    inputSchema: {
10
      type: 'object',
11
      properties: {
12
        title: { type: 'string' },
13
        subtitle: { type: 'string' },
14
        style: { type: 'string', enum: ['modern', 'classic', 'minimal'] },
15
        dimensions: { 
16
          type: 'object',
17
          properties: {
18
            width: { type: 'number', default: 1200 },
19
            height: { type: 'number', default: 630 }
20
          }
21
        }
22
      },
23
      required: ['title']
24
    }
25
  },
26
  {
27
    name: 'generate_color_palette',
28
    description: 'Generate a color palette for a brand',
29
    inputSchema: {
30
      type: 'object',
31
      properties: {
32
        base_color: { type: 'string' },
33
        palette_type: { type: 'string', enum: ['complementary', 'analogous', 'triadic'] }
34
      }
35
    }
36
  }
37
];
38
39
// Implementation similar to other servers...

 1
# agents/researcher/researcher.py
 2
import asyncio
 3
from typing import Dict, Any
 4
from hermes_agent import Agent
 5
 6
class ResearcherAgent(Agent):
 7
    def __init__(self):
 8
        super().__init__(name="researcher", description="Gathers and analyzes information")
 9
        
10
    async def process_request(self, task: Dict[str, Any]) -> Dict[str, Any]:
11
        """Process a research request"""
12
        
13
        # Use MCP tools for research
14
        research_results = await self.use_mcp_tool(
15
            server="research-server",
16
            tool="search_web",
17
            arguments={"query": task["topic"], "limit": 10}
18
        )
19
        
20
        analysis = await self.use_mcp_tool(
21
            server="research-server", 
22
            tool="analyze_trends",
23
            arguments={"data": research_results}
24
        )
25
        
26
        summary = await self.use_mcp_tool(
27
            server="research-server",
28
            tool="summarize_content",
29
            arguments={"text": analysis}
30
        )
31
        
32
        return {
33
            "status": "completed",
34
            "research_data": research_results,
35
            "analysis": analysis,
36
            "summary": summary,
37
            "recommendations": self.generate_recommendations(summary)
38
        }
39
    
40
    def generate_recommendations(self, summary: str) -> List[str]:
41
        """Generate content recommendations based on research"""
42
        # Implementation logic here
43
        return ["Focus on recent trends", "Include statistical data", "Add expert quotes"]

 1
# agents/writer/writer.py
 2
from hermes_agent import Agent
 3
 4
class WriterAgent(Agent):
 5
    def __init__(self):
 6
        super().__init__(name="writer", description="Creates compelling content")
 7
        
 8
    async def write_article(self, research_data: Dict[str, Any]) -> str:
 9
        """Write an article based on research"""
10
        
11
        article = await self.use_mcp_tool(
12
            server="writing-server",
13
            tool="generate_article",
14
            arguments={
15
                "topic": research_data["topic"],
16
                "length": "medium",
17
                "tone": "persuasive"
18
            }
19
        )
20
        
21
        # Improve the article
22
        improved_article = await self.use_mcp_tool(
23
            server="writing-server",
24
            tool="improve_text",
25
            arguments={
26
                "text": article,
27
                "goal": "engagement"
28
            }
29
        )
30
        
31
        return improved_article

 1
# agents/designer/designer.py
 2
from hermes_agent import Agent
 3
 4
class DesignerAgent(Agent):
 5
    def __init__(self):
 6
        super().__init__(name="designer", description="Creates visual content")
 7
        
 8
    async def create_designs(self, content_data: Dict[str, Any]) -> Dict[str, Any]:
 9
        """Create designs for content"""
10
        
11
        banner = await self.use_mcp_tool(
12
            server="design-server",
13
            tool="create_banner",
14
            arguments={
15
                "title": content_data["title"],
16
                "subtitle": content_data["subtitle"],
17
                "style": "modern",
18
                "dimensions": {"width": 1200, "height": 630}
19
            }
20
        )
21
        
22
        palette = await self.use_mcp_tool(
23
            server="design-server",
24
            tool="generate_color_palette",
25
            arguments={
26
                "base_color": "#3B82F6",
27
                "palette_type": "complementary"
28
            }
29
        )
30
        
31
        return {
32
            "banner": banner,
33
            "color_palette": palette,
34
            "design_recommendations": self.generate_design_tips()
35
        }

 1
# coordination/message_protocol.py
 2
from dataclasses import dataclass
 3
from typing import Dict, Any, Optional
 4
from datetime import datetime
 5
import json
 6
 7
@dataclass
 8
class AgentMessage:
 9
    message_id: str
10
    sender: str
11
    recipient: str
12
    message_type: str  # "task_request", "task_response", "status_update", "error"
13
    timestamp: datetime
14
    content: Dict[str, Any]
15
    priority: int = 1
16
    
17
    def to_json(self) -> str:
18
        return json.dumps({
19
            "message_id": self.message_id,
20
            "sender": self.sender,
21
            "recipient": self.recipient,
22
            "message_type": self.message_type,
23
            "timestamp": self.timestamp.isoformat(),
24
            "content": self.content,
25
            "priority": self.priority
26
        })
27
    
28
    @classmethod
29
    def from_json(cls, json_str: str) -> 'AgentMessage':
30
        data = json.loads(json_str)
31
        return cls(
32
            message_id=data["message_id"],
33
            sender=data["sender"],
34
            recipient=data["recipient"],
35
            message_type=data["message_type"],
36
            timestamp=datetime.fromisoformat(data["timestamp"]),
37
            content=data["content"],
38
            priority=data.get("priority", 1)
39
        )

 1
# coordination/message_broker.py
 2
import asyncio
 3
from typing import Dict, Set, Callable
 4
from .message_protocol import AgentMessage
 5
 6
class MessageBroker:
 7
    def __init__(self):
 8
        self.agents: Dict[str, asyncio.Queue] = {}
 9
        self.subscriptions: Dict[str, Set[str]] = {}
10
        self.message_handlers: Dict[str, Callable] = {}
11
        
12
    async def register_agent(self, agent_id: str) -> asyncio.Queue:
13
        """Register an agent with the broker"""
14
        if agent_id not in self.agents:
15
            self.agents[agent_id] = asyncio.Queue()
16
        return self.agents[agent_id]
17
    
18
    async def subscribe(self, agent_id: str, channel: str):
19
        """Subscribe agent to a channel"""
20
        if channel not in self.subscriptions:
21
            self.subscriptions[channel] = set()
22
        self.subscriptions[channel].add(agent_id)
23
    
24
    async def publish(self, message: AgentMessage):
25
        """Publish a message to recipients"""
26
        
27
        # Direct message
28
        if message.recipient in self.agents:
29
            await self.agents[message.recipient].put(message)
30
        
31
        # Channel subscriptions
32
        if message.recipient.startswith("channel:"):
33
            channel = message.recipient.split(":")[1]
34
            if channel in self.subscriptions:
35
                for subscriber in self.subscriptions[channel]:
36
                    if subscriber != message.sender:  # Don't send to self
37
                        await self.agents[subscriber].put(message)
38
    
39
    async def broadcast(self, message: AgentMessage):
40
        """Broadcast message to all agents"""
41
        for agent_id, queue in self.agents.items():
42
            if agent_id != message.sender:  # Don't send to self
43
                await queue.put(message)

  1
# coordination/orchestrator.py
  2
from hermes_agent import Agent
  3
from .message_broker import MessageBroker
  4
from .message_protocol import AgentMessage
  5
import uuid
  6
from datetime import datetime
  7
  8
class OrchestratorAgent(Agent):
  9
    def __init__(self, broker: MessageBroker):
 10
        super().__init__(name="orchestrator", description="Coordinates agent collaboration")
 11
        self.broker = broker
 12
        self.tasks: Dict[str, Dict] = {}  # task_id -> task_state
 13
        
 14
    async def coordinate_content_creation(self, topic: str) -> Dict[str, Any]:
 15
        """Coordinate creation of content on a topic"""
 16
        
 17
        task_id = str(uuid.uuid4())
 18
        self.tasks[task_id] = {
 19
            "topic": topic,
 20
            "status": "in_progress",
 21
            "agents_involved": [],
 22
            "results": {}
 23
        }
 24
        
 25
        # Step 1: Request research
 26
        research_request = AgentMessage(
 27
            message_id=str(uuid.uuid4()),
 28
            sender="orchestrator",
 29
            recipient="researcher",
 30
            message_type="task_request",
 31
            timestamp=datetime.now(),
 32
            content={
 33
                "task_id": task_id,
 34
                "action": "research",
 35
                "topic": topic,
 36
                "requirements": {
 37
                    "depth": "comprehensive",
 38
                    "sources": ["academic", "industry", "news"]
 39
                }
 40
            }
 41
        )
 42
        
 43
        await self.broker.publish(research_request)
 44
        
 45
        # Wait for research completion
 46
        research_result = await self.wait_for_task_completion(task_id, "research")
 47
        
 48
        # Step 2: Request writing based on research
 49
        writing_request = AgentMessage(
 50
            message_id=str(uuid.uuid4()),
 51
            sender="orchestrator",
 52
            recipient="writer",
 53
            message_type="task_request",
 54
            timestamp=datetime.now(),
 55
            content={
 56
                "task_id": task_id,
 57
                "action": "write",
 58
                "research_data": research_result,
 59
                "requirements": {
 60
                    "format": "blog_post",
 61
                    "target_audience": "technical",
 62
                    "word_count": 1500
 63
                }
 64
            }
 65
        )
 66
        
 67
        await self.broker.publish(writing_request)
 68
        
 69
        # Step 3: Request design based on content
 70
        writing_result = await self.wait_for_task_completion(task_id, "write")
 71
        
 72
        design_request = AgentMessage(
 73
            message_id=str(uuid.uuid4()),
 74
            sender="orchestrator",
 75
            recipient="designer",
 76
            message_type="task_request",
 77
            timestamp=datetime.now(),
 78
            content={
 79
                "task_id": task_id,
 80
                "action": "design",
 81
                "content_data": writing_result,
 82
                "requirements": {
 83
                    "output_formats": ["banner", "social_media", "featured_image"],
 84
                    "style": "modern_tech"
 85
                }
 86
            }
 87
        )
 88
        
 89
        await self.broker.publish(design_request)
 90
        
 91
        # Wait for all tasks to complete
 92
        design_result = await self.wait_for_task_completion(task_id, "design")
 93
        
 94
        # Aggregate results
 95
        final_result = {
 96
            "task_id": task_id,
 97
            "topic": topic,
 98
            "research": research_result,
 99
            "content": writing_result,
100
            "designs": design_result,
101
            "status": "completed",
102
            "completion_time": datetime.now().isoformat()
103
        }
104
        
105
        self.tasks[task_id]["status"] = "completed"
106
        self.tasks[task_id]["results"] = final_result
107
        
108
        return final_result
109
    
110
    async def wait_for_task_completion(self, task_id: str, action: str, timeout: int = 300):
111
        """Wait for a specific task action to complete"""
112
        # Implementation for waiting and receiving responses
113
        pass

 1
# config/ecosystem.yaml
 2
mcp_servers:
 3
  research:
 4
    command: "node"
 5
    args: ["./mcp-servers/research-server/index.js"]
 6
    env:
 7
      OPENAI_API_KEY: "${OPENAI_API_KEY}"
 8
      SERPER_API_KEY: "${SERPER_API_KEY}"
 9
    
10
  writing:
11
    command: "python"
12
    args: ["./mcp-servers/writing-server/server.py"]
13
    env:
14
      OPENAI_API_KEY: "${OPENAI_API_KEY}"
15
    
16
  design:
17
    command: "node"
18
    args: ["./mcp-servers/design-server/index.js"]
19
20
agents:
21
  researcher:
22
    class: "agents.researcher.ResearcherAgent"
23
    mcp_servers: ["research"]
24
    capabilities: ["web_research", "data_analysis", "trend_analysis"]
25
    
26
  writer:
27
    class: "agents.writer.WriterAgent"
28
    mcp_servers: ["writing"]
29
    capabilities: ["content_creation", "editing", "seo_optimization"]
30
    
31
  designer:
32
    class: "agents.designer.DesignerAgent"
33
    mcp_servers: ["design"]
34
    capabilities: ["graphic_design", "ui_design", "color_theory"]
35
    
36
  orchestrator:
37
    class: "coordination.orchestrator.OrchestratorAgent"
38
    dependencies: ["researcher", "writer", "designer"]
39
40
communication:
41
  broker_class: "coordination.message_broker.MessageBroker"
42
  protocols:
43
    - "direct_messaging"
44
    - "channel_broadcast"
45
    - "task_delegation"
46
  
47
  channels:
48
    research_updates: ["researcher", "orchestrator"]
49
    content_updates: ["writer", "orchestrator"]
50
    design_updates: ["designer", "orchestrator"]
51
    system_alerts: ["orchestrator", "researcher", "writer", "designer"]

 1
# main.py
 2
import asyncio
 3
import yaml
 4
from coordination.message_broker import MessageBroker
 5
from coordination.orchestrator import OrchestratorAgent
 6
from agents.researcher import ResearcherAgent
 7
from agents.writer import WriterAgent
 8
from agents.designer import DesignerAgent
 9
10
class ContentStudio:
11
    def __init__(self, config_path: str = "config/ecosystem.yaml"):
12
        with open(config_path, 'r') as f:
13
            self.config = yaml.safe_load(f)
14
        
15
        self.broker = MessageBroker()
16
        self.agents = {}
17
        self.mcp_servers = {}
18
        
19
    async def initialize(self):
20
        """Initialize the entire ecosystem"""
21
        
22
        # Start MCP servers
23
        await self.start_mcp_servers()
24
        
25
        # Initialize agents
26
        await self.initialize_agents()
27
        
28
        # Set up communication
29
        await self.setup_communication()
30
        
31
        print("Content Studio Ecosystem initialized successfully!")
32
    
33
    async def start_mcp_servers(self):
34
        """Start all configured MCP servers"""
35
        for server_name, server_config in self.config["mcp_servers"].items():
36
            print(f"Starting MCP server: {server_name}")
37
            # Implementation for starting MCP servers
38
            pass
39
    
40
    async def initialize_agents(self):
41
        """Initialize all agents"""
42
        for agent_name, agent_config in self.config["agents"].items():
43
            agent_class = self.import_class(agent_config["class"])
44
            
45
            if agent_name == "orchestrator":
46
                agent = agent_class(self.broker)
47
            else:
48
                agent = agent_class()
49
            
50
            self.agents[agent_name] = agent
51
            
52
            # Register agent with message broker
53
            await self.broker.register_agent(agent_name)
54
            
55
            print(f"Initialized agent: {agent_name}")
56
    
57
    async def setup_communication(self):
58
        """Set up communication channels"""
59
        for channel, subscribers in self.config["communication"]["channels"].items():
60
            for subscriber in subscribers:
61
                await self.broker.subscribe(subscriber, channel)
62
    
63
    async def create_content(self, topic: str) -> dict:
64
        """Create content on a given topic"""
65
        orchestrator = self.agents["orchestrator"]
66
        return await orchestrator.coordinate_content_creation(topic)
67
    
68
    def import_class(self, class_path: str):
69
        """Dynamically import a class"""
70
        module_path, class_name = class_path.rsplit('.', 1)
71
        module = __import__(module_path, fromlist=[class_name])
72
        return getattr(module, class_name)
73
74
async def main():
75
    studio = ContentStudio()
76
    await studio.initialize()
77
    
78
    # Example: Create content about "The Future of AI"
79
    result = await studio.create_content("The Future of Artificial Intelligence")
80
    
81
    print(f"Content creation completed!")
82
    print(f"Task ID: {result['task_id']}")
83
    print(f"Research summary: {result['research']['summary'][:200]}...")
84
    print(f"Content created: {len(result['content'])} characters")
85
    print(f"Designs generated: {len(result['designs'])}")
86
87
if __name__ == "__main__":
88
    asyncio.run(main())

 1
# tests/test_ecosystem.py
 2
import pytest
 3
import asyncio
 4
from main import ContentStudio
 5
 6
@pytest.mark.asyncio
 7
async def test_content_creation():
 8
    studio = ContentStudio("config/test_ecosystem.yaml")
 9
    await studio.initialize()
10
    
11
    result = await studio.create_content("Test Topic")
12
    
13
    assert result["status"] == "completed"
14
    assert "research" in result
15
    assert "content" in result
16
    assert "designs" in result
17
    assert result["topic"] == "Test Topic"

 1
# Dockerfile
 2
FROM python:3.11-slim
 3
 4
WORKDIR /app
 5
 6
# Install system dependencies
 7
RUN apt-get update && apt-get install -y \
 8
    nodejs \
 9
    npm \
10
    && rm -rf /var/lib/apt/lists/*
11
12
# Copy project files
13
COPY . .
14
15
# Install Python dependencies
16
RUN pip install --no-cache-dir \
17
    hermes-agent \
18
    mcp \
19
    pyyaml \
20
    pytest-asyncio
21
22
# Install Node.js dependencies
23
RUN npm install --prefix ./mcp-servers/research-server \
24
    @modelcontextprotocol/sdk
25
26
# Create non-root user
27
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
28
USER appuser
29
30
# Run the application
31
CMD ["python", "main.py"]

 1
# kubernetes/deployment.yaml
 2
apiVersion: apps/v1
 3
kind: Deployment
 4
metadata:
 5
  name: content-studio
 6
spec:
 7
  replicas: 3
 8
  selector:
 9
    matchLabels:
10
      app: content-studio
11
  template:
12
    metadata:
13
      labels:
14
        app: content-studio
15
    spec:
16
      containers:
17
      - name: studio
18
        image: content-studio:latest
19
        ports:
20
        - containerPort: 8080
21
        env:
22
        - name: OPENAI_API_KEY
23
          valueFrom:
24
            secretKeyRef:
25
              name: api-secrets
26
              key: openai-api-key
27
        resources:
28
          requests:
29
            memory: "512Mi"
30
            cpu: "500m"
31
          limits:
32
            memory: "1Gi"
33
            cpu: "1"

 1
# monitoring/metrics.py
 2
from prometheus_client import Counter, Histogram, Gauge
 3
import time
 4
 5
# Define metrics
 6
agent_requests_total = Counter('agent_requests_total', 'Total agent requests', ['agent', 'status'])
 7
task_duration_seconds = Histogram('task_duration_seconds', 'Task duration in seconds', ['task_type'])
 8
active_agents = Gauge('active_agents', 'Number of active agents')
 9
mcp_tool_calls_total = Counter('mcp_tool_calls_total', 'Total MCP tool calls', ['server', 'tool', 'status'])
10
11
def track_agent_request(agent_name: str, success: bool):
12
    agent_requests_total.labels(agent=agent_name, status='success' if success else 'failure').inc()
13
14
def track_task_duration(task_type: str, duration: float):
15
    task_duration_seconds.labels(task_type=task_type).observe(duration)
16
17
def track_mcp_tool_call(server: str, tool: str, success: bool):
18
    mcp_tool_calls_total.labels(server=server, tool=tool, status='success' if success else 'failure').inc()

 1
# monitoring/logging_config.py
 2
import logging
 3
import json
 4
from datetime import datetime
 5
 6
class StructuredLogger:
 7
    def __init__(self, name: str):
 8
        self.logger = logging.getLogger(name)
 9
        
10
    def log_agent_interaction(self, sender: str, recipient: str, message_type: str, content: dict):
11
        log_entry = {
12
            "timestamp": datetime.now().isoformat(),
13
            "level": "INFO",
14
            "type": "agent_interaction",
15
            "sender": sender,
16
            "recipient": recipient,
17
            "message_type": message_type,
18
            "content": content
19
        }
20
        self.logger.info(json.dumps(log_entry))
21
    
22
    def log_mcp_tool_call(self, server: str, tool: str, parameters: dict, result: dict, duration: float):
23
        log_entry = {
24
            "timestamp": datetime.now().isoformat(),
25
            "level": "INFO",
26
            "type": "mcp_tool_call",
27
            "server": server,
28
            "tool": tool,
29
            "parameters": parameters,
30
            "result": result,
31
            "duration_seconds": duration
32
        }
33
        self.logger.info(json.dumps(log_entry))