Build Your Own MCP Server

What You’ll Learn

This guide shows you how to build your own MCP (Model Context Protocol) server to integrate custom APIs, internal tools, or proprietary systems with Junis. You’ll learn:

MCP protocol fundamentals
Building MCP servers in Python and Node.js
Implementing tools and resources
Authentication and security
Deploying and testing your MCP server
Registering with Junis

Prerequisites:

Familiarity with REST APIs
Python (3.9+) or Node.js (18+) knowledge
Understanding of MCP Overview
Docker (optional, for deployment)

Why Build a Custom MCP Server?

Internal APIs

Connect agents to your company’s proprietary APIs, microservices, or legacy systems.

Unique Business Logic

Implement custom workflows, calculations, or processes specific to your business.

Data Integration

Aggregate data from multiple internal sources into a unified interface.

Security & Compliance

Keep sensitive data within your infrastructure without exposing it to third-party MCPs.

MCP Protocol Basics

MCP Architecture

MCP Request/Response Flow

1. Tool Discovery:

// Request from Junis
{
  "method": "list_tools",
  "params": {}
}

// Response from your MCP server
{
  "tools": [
    {
      "name": "get_customer_data",
      "description": "Retrieve customer information by ID",
      "parameters": {
        "customer_id": {"type": "string", "required": true}
      }
    }
  ]
}

2. Tool Execution:

// Request from Junis
{
  "method": "call_tool",
  "params": {
    "tool_name": "get_customer_data",
    "arguments": {"customer_id": "CUST-12345"}
  }
}

// Response from your MCP server
{
  "result": {
    "customer_id": "CUST-12345",
    "name": "Acme Corp",
    "tier": "Enterprise",
    "mrr": 5000
  }
}

Quick Start: Python MCP Server

Project Setup

Create Project Structure

mkdir my-mcp-server
cd my-mcp-server

# Create files
touch server.py requirements.txt README.md

# Project structure
my-mcp-server/
├── server.py
├── requirements.txt
├── tools/
│   ├── __init__.py
│   └── customer_tools.py
└── README.md

Install Dependencies

requirements.txt:

fastapi==0.104.1
uvicorn==0.24.0
pydantic==2.5.0
httpx==0.25.1
python-dotenv==1.0.0

Install:

pip install -r requirements.txt

Create MCP Server

server.py:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, Dict, Any, List
import os

app = FastAPI(title="My Custom MCP Server")

# Tool definitions
TOOLS = [
    {
        "name": "get_customer_data",
        "description": "Retrieve customer information by ID",
        "parameters": {
            "type": "object",
            "properties": {
                "customer_id": {
                    "type": "string",
                    "description": "Customer ID (e.g., CUST-12345)"
                }
            },
            "required": ["customer_id"]
        }
    },
    {
        "name": "calculate_discount",
        "description": "Calculate discount based on customer tier",
        "parameters": {
            "type": "object",
            "properties": {
                "amount": {"type": "number"},
                "tier": {"type": "string", "enum": ["basic", "premium", "enterprise"]}
            },
            "required": ["amount", "tier"]
        }
    }
]

class MCPRequest(BaseModel):
    method: str
    params: Optional[Dict[str, Any]] = {}

@app.post("/")
async def handle_mcp_request(request: MCPRequest):
    """Main MCP endpoint"""
    if request.method == "list_tools":
        return {"tools": TOOLS}

    elif request.method == "call_tool":
        tool_name = request.params.get("tool_name")
        arguments = request.params.get("arguments", {})

        if tool_name == "get_customer_data":
            return await get_customer_data(arguments["customer_id"])

        elif tool_name == "calculate_discount":
            return await calculate_discount(
                arguments["amount"],
                arguments["tier"]
            )

        else:
            raise HTTPException(status_code=404, detail=f"Tool {tool_name} not found")

    else:
        raise HTTPException(status_code=400, detail=f"Unknown method: {request.method}")

async def get_customer_data(customer_id: str):
    """Tool implementation: Get customer data"""
    # TODO: Replace with your actual API call
    mock_data = {
        "CUST-12345": {
            "customer_id": "CUST-12345",
            "name": "Acme Corp",
            "tier": "enterprise",
            "mrr": 5000,
            "industry": "Technology"
        }
    }

    if customer_id in mock_data:
        return {"result": mock_data[customer_id]}
    else:
        raise HTTPException(status_code=404, detail="Customer not found")

async def calculate_discount(amount: float, tier: str):
    """Tool implementation: Calculate discount"""
    discount_rates = {
        "basic": 0.05,
        "premium": 0.10,
        "enterprise": 0.20
    }

    discount = amount * discount_rates.get(tier, 0)
    final_amount = amount - discount

    return {
        "result": {
            "original_amount": amount,
            "discount_rate": discount_rates[tier],
            "discount": discount,
            "final_amount": final_amount
        }
    }

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "tools_count": len(TOOLS)}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8766)

Run the Server

python server.py

Server will start on: http://localhost:8766

Test the Server

# List tools
curl -X POST http://localhost:8766/ \
  -H "Content-Type: application/json" \
  -d '{"method": "list_tools", "params": {}}'

# Call a tool
curl -X POST http://localhost:8766/ \
  -H "Content-Type: application/json" \
  -d '{
    "method": "call_tool",
    "params": {
      "tool_name": "get_customer_data",
      "arguments": {"customer_id": "CUST-12345"}
    }
  }'

Quick Start: Node.js MCP Server

Project Setup

Initialize Project

mkdir my-mcp-server-js
cd my-mcp-server-js
npm init -y
npm install express body-parser axios dotenv

Create Server

server.js:

const express = require('express');
const bodyParser = require('body-parser');
require('dotenv').config();

const app = express();
app.use(bodyParser.json());

// Tool definitions
const TOOLS = [
  {
    name: 'get_product_info',
    description: 'Retrieve product information by SKU',
    parameters: {
      type: 'object',
      properties: {
        sku: {
          type: 'string',
          description: 'Product SKU code'
        }
      },
      required: ['sku']
    }
  }
];

// Main MCP endpoint
app.post('/', async (req, res) => {
  const { method, params } = req.body;

  try {
    if (method === 'list_tools') {
      res.json({ tools: TOOLS });
    } else if (method === 'call_tool') {
      const { tool_name, arguments: args } = params;

      if (tool_name === 'get_product_info') {
        const result = await getProductInfo(args.sku);
        res.json({ result });
      } else {
        res.status(404).json({ error: `Tool ${tool_name} not found` });
      }
    } else {
      res.status(400).json({ error: `Unknown method: ${method}` });
    }
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

// Tool implementation
async function getProductInfo(sku) {
  // TODO: Replace with your actual API call
  const mockData = {
    'SKU-001': {
      sku: 'SKU-001',
      name: 'Premium Widget',
      price: 99.99,
      stock: 150,
      category: 'Electronics'
    }
  };

  if (mockData[sku]) {
    return mockData[sku];
  } else {
    throw new Error('Product not found');
  }
}

// Health check
app.get('/health', (req, res) => {
  res.json({ status: 'healthy', tools_count: TOOLS.length });
});

const PORT = process.env.PORT || 8766;
app.listen(PORT, () => {
  console.log(`MCP Server running on port ${PORT}`);
});

Adding Authentication

API Key Authentication

Python (FastAPI)
Node.js (Express)

from fastapi import FastAPI, HTTPException, Header
from typing import Optional

app = FastAPI()

# Simple API key validation
VALID_API_KEYS = {
    "sk-test-123": "TestOrg",
    "sk-prod-456": "ProdOrg"
}

async def verify_api_key(authorization: Optional[str] = Header(None)):
    if not authorization:
        raise HTTPException(status_code=401, detail="Missing API key")

    # Extract key from "Bearer <key>"
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid authorization format")

    api_key = authorization.replace("Bearer ", "")

    if api_key not in VALID_API_KEYS:
        raise HTTPException(status_code=401, detail="Invalid API key")

    return VALID_API_KEYS[api_key]  # Returns organization name

@app.post("/")
async def handle_mcp_request(
    request: MCPRequest,
    organization: str = Depends(verify_api_key)
):
    # Use 'organization' to scope data access
    print(f"Request from organization: {organization}")
    # ... rest of handler

const VALID_API_KEYS = {
  'sk-test-123': 'TestOrg',
  'sk-prod-456': 'ProdOrg'
};

// Authentication middleware
function verifyApiKey(req, res, next) {
  const authHeader = req.headers.authorization;

  if (!authHeader) {
    return res.status(401).json({ error: 'Missing API key' });
  }

  if (!authHeader.startsWith('Bearer ')) {
    return res.status(401).json({ error: 'Invalid authorization format' });
  }

  const apiKey = authHeader.replace('Bearer ', '');

  if (!VALID_API_KEYS[apiKey]) {
    return res.status(401).json({ error: 'Invalid API key' });
  }

  req.organization = VALID_API_KEYS[apiKey];
  next();
}

// Apply middleware to all routes
app.use(verifyApiKey);

app.post('/', async (req, res) => {
  console.log(`Request from organization: ${req.organization}`);
  // ... rest of handler
});

OAuth 2.0 Authentication

For production systems, implement OAuth 2.0:

from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

async def verify_oauth_token(
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    token = credentials.credentials

    try:
        # Verify JWT token
        payload = jwt.decode(
            token,
            os.getenv("JWT_SECRET"),
            algorithms=["HS256"]
        )
        return payload  # Contains user/org info
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

Advanced Tool Implementation

Streaming Responses

For long-running operations, implement streaming:

from fastapi.responses import StreamingResponse
import asyncio

async def stream_large_dataset():
    """Stream data incrementally"""
    for chunk in fetch_data_chunks():
        yield json.dumps(chunk) + "\n"
        await asyncio.sleep(0.1)  # Simulate processing

@app.post("/stream-tool")
async def handle_stream_tool():
    return StreamingResponse(
        stream_large_dataset(),
        media_type="application/x-ndjson"
    )

Error Handling

Implement comprehensive error handling:

from enum import Enum

class MCPErrorCode(str, Enum):
    TOOL_NOT_FOUND = "tool_not_found"
    INVALID_PARAMETERS = "invalid_parameters"
    INTERNAL_ERROR = "internal_error"
    RATE_LIMIT_EXCEEDED = "rate_limit_exceeded"

class MCPError(Exception):
    def __init__(self, code: MCPErrorCode, message: str, details: dict = None):
        self.code = code
        self.message = message
        self.details = details or {}

@app.exception_handler(MCPError)
async def mcp_error_handler(request, exc: MCPError):
    return JSONResponse(
        status_code=400,
        content={
            "error": {
                "code": exc.code,
                "message": exc.message,
                "details": exc.details
            }
        }
    )

Parameter Validation

Use Pydantic for strict parameter validation:

from pydantic import BaseModel, Field, validator

class GetCustomerParams(BaseModel):
    customer_id: str = Field(..., regex="^CUST-\d{5}$")

    @validator('customer_id')
    def validate_customer_id(cls, v):
        if not v.startswith("CUST-"):
            raise ValueError("Customer ID must start with CUST-")
        return v

async def get_customer_data(params: GetCustomerParams):
    # params.customer_id is validated
    pass

Deployment

Docker Deployment

Dockerfile:

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8766

CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8766"]

Build and Run:

docker build -t my-mcp-server .
docker run -d -p 8766:8766 --name mcp-server my-mcp-server

Production Considerations

✅ Production Checklist:

HTTPS/TLS enabled (use reverse proxy like Nginx)
Rate limiting implemented
Logging and monitoring configured
Error tracking (Sentry, Rollbar)
Health check endpoint exposed
Secrets managed via environment variables
API versioning implemented
Documentation generated (OpenAPI/Swagger)

Registering with Junis

Step 1: Deploy Your MCP Server

Ensure your server is publicly accessible or within your VPN:

https://mcp.yourcompany.com/

Step 2: Add Platform in Junis

Admin Panel:

Go to Team > MCP
Click “Connect”
Fill in details:
- Platform Name: Your Company MCP
- MCP Server URL: https://mcp.yourcompany.com/
- Transport Type: Streamable HTTP
- Auth Type: API Key or OAuth2

Step 3: Add Credentials

Organization-Level:

Admin > MCP > [Your Platform] > Add Auth
Enter API key or OAuth credentials

User-Level (if applicable):

User Settings > MCP Credentials > [Your Platform]
Complete authentication flow

Step 4: Enable for Agents

Admin > Agents > [Your Agent]
MCP Platforms > Check your custom platform
Save

Step 5: Test

Ask your agent:

"Use [your tool name] to [perform action]"

Examples & Templates

Example 1: Internal CRM Integration

Use Case: Query customer data from internal CRM Tools:

crm_search_customers - Search by name, email, or company
crm_get_customer - Get full customer profile
crm_get_interactions - List recent interactions
crm_create_note - Add note to customer record

Reference: See MCP Server Examples for similar implementations

Example 2: Data Warehouse Query

Use Case: Run analytics queries on data warehouse Tools:

dw_execute_query - Run SQL query
dw_list_tables - List available tables
dw_get_schema - Get table schema
dw_export_results - Export results to CSV

Reference: See PostgreSQL MCP Implementation for reference

Example 3: Workflow Automation

Use Case: Trigger internal workflows Tools:

workflow_start - Start workflow with parameters
workflow_status - Check workflow status
workflow_cancel - Cancel running workflow
workflow_list_templates - List available workflow templates

Reference: Implement based on your internal workflow engine API

Best Practices

✅ DO:

Document all tools with clear descriptions
Implement comprehensive error handling
Use typed parameters (JSON Schema)
Log all requests for debugging
Implement rate limiting per API key
Version your MCP API (/v1/, /v2/)
Provide health check endpoint
Test with realistic data

❌ DON’T:

Expose sensitive data without authentication
Return unbounded result sets (always paginate)
Make blocking calls (use async)
Trust input without validation
Log sensitive data (passwords, API keys)
Deploy without monitoring
Skip documentation

What’s Next?

MCP Protocol Spec

Deep dive into MCP protocol details

MCP Overview

See how other platforms implement MCP

GitHub MCP

Connect to GitHub repositories

Community MCPs

Browse open-source MCP servers

Additional Resources

MCP Protocol Spec: https://modelcontextprotocol.io/
Example Servers: https://github.com/modelcontextprotocol/servers
FastAPI Documentation: https://fastapi.tiangolo.com/
Express.js Documentation: https://expressjs.com/

Need Help? Check the MCP Protocol Specification and MCP Server Examples for implementation guidance and best practices.

​What You’ll Learn

​Why Build a Custom MCP Server?

Internal APIs

Unique Business Logic

Data Integration

Security & Compliance

​MCP Protocol Basics

​MCP Architecture

​MCP Request/Response Flow

​Quick Start: Python MCP Server

​Project Setup

​Quick Start: Node.js MCP Server

​Project Setup

​Adding Authentication

​API Key Authentication

​OAuth 2.0 Authentication

​Advanced Tool Implementation

​Streaming Responses

​Error Handling

​Parameter Validation

​Deployment

​Docker Deployment

​Production Considerations

​Registering with Junis

​Step 1: Deploy Your MCP Server

​Step 2: Add Platform in Junis

​Step 3: Add Credentials

​Step 4: Enable for Agents

​Step 5: Test

​Examples & Templates

​Example 1: Internal CRM Integration

​Example 2: Data Warehouse Query

​Example 3: Workflow Automation

​Best Practices

​What’s Next?

MCP Protocol Spec

MCP Overview

GitHub MCP

Community MCPs

​Additional Resources

What You’ll Learn

Why Build a Custom MCP Server?

MCP Protocol Basics

MCP Architecture

MCP Request/Response Flow

Quick Start: Python MCP Server

Project Setup

Quick Start: Node.js MCP Server

Project Setup

Adding Authentication

API Key Authentication

OAuth 2.0 Authentication

Advanced Tool Implementation

Streaming Responses

Error Handling

Parameter Validation

Deployment

Docker Deployment

Production Considerations

Registering with Junis

Step 1: Deploy Your MCP Server

Step 2: Add Platform in Junis

Step 3: Add Credentials

Step 4: Enable for Agents

Step 5: Test

Examples & Templates

Example 1: Internal CRM Integration

Example 2: Data Warehouse Query

Example 3: Workflow Automation

Best Practices

What’s Next?

Additional Resources