MCP Server Development
Last updated
{
"status": "ok",
"uptime": 102934,
"timestamp": "2025-06-30T12:34:56Z"
}{
"functions": [
{
"name": "extract_entities",
"description": "Extracts named entities from a string",
"inputs": ["text"],
"outputs": ["entities"]
},
{
"name": "get_exchange_rate",
"description": "Returns current exchange rate between two currencies",
"inputs": ["from", "to"],
"outputs": ["rate"]
}
]
}Content-Type: application/json{
"function": "get_exchange_rate",
"input": {
"from": "USD",
"to": "EUR"
}
}{
"output": {
"rate": 0.9254
}
}const express = require('express');
const app = express();
// Middleware
app.use(express.json());
app.use(require('cors')());
// Server configuration
const SERVER_CONFIG = {
name: 'My MCP Server',
version: '1.0.0',
mcpVersion: '1.0.0',
description: 'Description of your server'
};
// Function registry
const functions = new Map();
// Register a function
function registerFunction(name, handler, schema) {
functions.set(name, { handler, schema });
}
// Health endpoint
app.get('/health', (req, res) => {
res.json({
status: 'healthy',
timestamp: new Date().toISOString(),
version: SERVER_CONFIG.version,
mcpVersion: SERVER_CONFIG.mcpVersion
});
});
// Capabilities endpoint
app.get('/capabilities', (req, res) => {
const functionList = Array.from(functions.entries()).map(([name, func]) => ({
name,
description: func.schema.description,
parameters: func.schema.parameters,
returns: func.schema.returns
}));
res.json({
mcpVersion: SERVER_CONFIG.mcpVersion,
server: {
name: SERVER_CONFIG.name,
version: SERVER_CONFIG.version,
description: SERVER_CONFIG.description
},
functions: functionList
});
});
// Execute endpoint
app.post('/execute', async (req, res) => {
const { function: functionName, parameters, requestId } = req.body;
try {
const func = functions.get(functionName);
if (!func) {
return res.status(400).json({
error: {
code: 'FUNCTION_NOT_FOUND',
message: `Function '${functionName}' not found`
},
requestId
});
}
const startTime = Date.now();
const result = await func.handler(parameters);
const executionTime = Date.now() - startTime;
res.json({
requestId,
result,
executionTime
});
} catch (error) {
res.status(500).json({
error: {
code: 'EXECUTION_ERROR',
message: error.message
},
requestId
});
}
});
// Example function registration
registerFunction('getPrice', async (params) => {
// Your function logic here
return {
price: 2345.67,
timestamp: new Date().toISOString(),
symbol: params.symbol
};
}, {
description: 'Get current price for a trading pair',
parameters: {
type: 'object',
properties: {
symbol: { type: 'string', description: 'Trading pair symbol' }
},
required: ['symbol']
},
returns: {
type: 'object',
properties: {
price: { type: 'number' },
timestamp: { type: 'string' },
symbol: { type: 'string' }
}
}
});
const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
console.log(`MCP Server running on port ${PORT}`);
});from flask import Flask, request, jsonify
from datetime import datetime
import time
app = Flask(__name__)
# Server configuration
SERVER_CONFIG = {
'name': 'My MCP Server',
'version': '1.0.0',
'mcpVersion': '1.0.0',
'description': 'Description of your server'
}
# Function registry
functions = {}
def register_function(name, handler, schema):
functions[name] = {'handler': handler, 'schema': schema}
@app.route('/health', methods=['GET'])
def health():
return jsonify({
'status': 'healthy',
'timestamp': datetime.utcnow().isoformat() + 'Z',
'version': SERVER_CONFIG['version'],
'mcpVersion': SERVER_CONFIG['mcpVersion']
})
@app.route('/capabilities', methods=['GET'])
def capabilities():
function_list = []
for name, func in functions.items():
function_list.append({
'name': name,
'description': func['schema']['description'],
'parameters': func['schema']['parameters'],
'returns': func['schema']['returns']
})
return jsonify({
'mcpVersion': SERVER_CONFIG['mcpVersion'],
'server': {
'name': SERVER_CONFIG['name'],
'version': SERVER_CONFIG['version'],
'description': SERVER_CONFIG['description']
},
'functions': function_list
})
@app.route('/execute', methods=['POST'])
def execute():
data = request.get_json()
function_name = data.get('function')
parameters = data.get('parameters', {})
request_id = data.get('requestId')
try:
if function_name not in functions:
return jsonify({
'error': {
'code': 'FUNCTION_NOT_FOUND',
'message': f"Function '{function_name}' not found"
},
'requestId': request_id
}), 400
start_time = time.time()
result = functions[function_name]['handler'](parameters)
execution_time = int((time.time() - start_time) * 1000)
return jsonify({
'requestId': request_id,
'result': result,
'executionTime': execution_time
})
except Exception as e:
return jsonify({
'error': {
'code': 'EXECUTION_ERROR',
'message': str(e)
},
'requestId': request_id
}), 500
# Example function
def get_price(params):
return {
'price': 2345.67,
'timestamp': datetime.utcnow().isoformat() + 'Z',
'symbol': params.get('symbol')
}
# Register the function
register_function('getPrice', get_price, {
'description': 'Get current price for a trading pair',
'parameters': {
'type': 'object',
'properties': {
'symbol': {'type': 'string', 'description': 'Trading pair symbol'}
},
'required': ['symbol']
},
'returns': {
'type': 'object',
'properties': {
'price': {'type': 'number'},
'timestamp': {'type': 'string'},
'symbol': {'type': 'string'}
}
}
})
if __name__ == '__main__':
app.run(debug=True, port=3000)# Install testing tools
npm install -g @tairon/test-suite
# Test your server
tairon-test http://localhost:3000 --comprehensive
# Example output:
# Health check endpoint
# Capabilities discovery
# Function execution
# Error handling
# Schema validation
# Rate limiting not implemented
# Security scan passed