MCP AI Agent: A Deep Dive into Model Context Protocol

1from http.server import BaseHTTPRequestHandler, HTTPServer
2import json
3
4class MCPHandler(BaseHTTPRequestHandler):
5    def do_POST(self):
6        content_length = int(self.headers['Content-Length'])
7        post_data = self.rfile.read(content_length)
8        input_data = json.loads(post_data.decode('utf-8'))
9
10        # Process the input_data (MCP request) and generate a response
11        response_data = {"result": "Processed: " + input_data.get("command", "")}
12
13        self.send_response(200)
14        self.send_header('Content-type', 'application/json')
15        self.end_headers()
16        self.wfile.write(json.dumps(response_data).encode('utf-8'))
17
18def run(server_class=HTTPServer, handler_class=MCPHandler, port=8000):
19    server_address = ('', port)
20    httpd = server_class(server_address, handler_class)
21    print(f'Starting MCP server on port {port}')
22    httpd.serve_forever()
23
24if __name__ == '__main__':
25    run()
26

1import subprocess
2import json
3
4def send_mcp_request(command, server_path):
5    process = subprocess.Popen(
6        [server_path],
7        stdin=subprocess.PIPE,
8        stdout=subprocess.PIPE,
9        stderr=subprocess.PIPE,
10        text=True
11    )
12    request_data = json.dumps({"command": command})
13    stdout, stderr = process.communicate(input=request_data)
14    
15    if stderr:
16        print(f"Error: {stderr}")
17        return None
18    
19    try:
20        return json.loads(stdout)
21    except json.JSONDecodeError:
22        print(f"Invalid JSON response: {stdout}")
23        return None
24
25# Example usage
26server_path = "./mcp_server.py" # Replace with the actual path to your MCP server script
27command = "summarize this document"
28response = send_mcp_request(command, server_path)
29
30if response:
31    print(f"Response from MCP server: {response}")
32

1from http.server import BaseHTTPRequestHandler, HTTPServer
2import time
3import json
4
5class SSEHandler(BaseHTTPRequestHandler):
6    def do_GET(self):
7        if self.path == '/events':
8            self.send_response(200)
9            self.send_header('Content-type', 'text/event-stream')
10            self.send_header('Cache-Control', 'no-cache')
11            self.end_headers()
12
13            try:
14                while True:
15                    # Generate some data (replace with your actual data source)
16                    data = {"message": f"Server time: {time.strftime('%H:%M:%S')}"}
17                    payload = f"data: {json.dumps(data)}
18
19"
20                    self.wfile.write(payload.encode('utf-8'))
21                    self.wfile.flush()
22                    time.sleep(1)
23            except BrokenPipeError:
24                print("Client disconnected")
25        else:
26            self.send_response(404)
27            self.end_headers()
28
29def run(server_class=HTTPServer, handler_class=SSEHandler, port=8000):
30    server_address = ('', port)
31    httpd = server_class(server_address, handler_class)
32    print(f'Starting SSE MCP server on port {port}')
33    httpd.serve_forever()
34
35if __name__ == '__main__':
36    run()
37

1from flask import Flask, request, jsonify
2
3app = Flask(__name__)
4
5@app.route('/mcp', methods=['POST'])
6def mcp_endpoint():
7    data = request.get_json()
8    command = data.get('command')
9
10    # Process the command (e.g., call a function, query a database)
11    result = process_command(command)
12
13    response = {'result': result}
14    return jsonify(response)
15
16def process_command(command):
17    # Replace with your actual command processing logic
18    return f'Processed command: {command}'
19
20if __name__ == '__main__':
21    app.run(debug=True, port=5000)
22

1import requests
2import json
3
4server_url = 'http://localhost:5000/mcp'
5
6request_data = {'command': 'generate a summary'}
7
8response = requests.post(server_url, json=request_data)
9
10if response.status_code == 200:
11    result = response.json()['result']
12    print(f'Result from server: {result}')
13else:
14    print(f'Error: {response.status_code} - {response.text}')
15

1# Server 1: Extracts entities from text
2def extract_entities(text):
3    # ... (Entity extraction logic)
4    return {'entities': ['person: John Doe', 'location: New York']}
5
6# Server 2: Summarizes information about entities
7def summarize_entities(entities):
8    # ... (Summarization logic)
9    return 'John Doe is a person from New York.'
10
11# Client: Chains the servers
12def process_text(text):
13    entities = extract_entities(text)
14    summary = summarize_entities(entities)
15    return summary
16