> ## Documentation Index
> Fetch the complete documentation index at: https://docs.junis.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Build Your Own MCP Server

> Create custom MCP integrations for your proprietary APIs and internal systems

## 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

<Info>
  **Prerequisites**:

  * Familiarity with REST APIs
  * Python (3.9+) or Node.js (18+) knowledge
  * Understanding of [MCP Overview](/guides/mcp/overview)
  * Docker (optional, for deployment)
</Info>

***

## Why Build a Custom MCP Server?

<CardGroup cols={2}>
  <Card title="Internal APIs" icon="building">
    Connect agents to your company's proprietary APIs, microservices, or legacy systems.
  </Card>

  <Card title="Unique Business Logic" icon="brain">
    Implement custom workflows, calculations, or processes specific to your business.
  </Card>

  <Card title="Data Integration" icon="database">
    Aggregate data from multiple internal sources into a unified interface.
  </Card>

  <Card title="Security & Compliance" icon="shield">
    Keep sensitive data within your infrastructure without exposing it to third-party MCPs.
  </Card>
</CardGroup>

***

## MCP Protocol Basics

### MCP Architecture

```mermaid theme={null}
graph LR
    A[Junis Agent] -->|MCP Protocol| B[Your MCP Server]
    B -->|Internal API| C[Your Backend]
    B -->|Database Query| D[(Your Database)]
    B -->|External API| E[Third-Party Service]
```

### MCP Request/Response Flow

**1. Tool Discovery**:

```json theme={null}
// 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**:

```json theme={null}
// 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

<Steps>
  <Step title="Create Project Structure">
    ```bash theme={null}
    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
    ```
  </Step>

  <Step title="Install Dependencies">
    **requirements.txt**:

    ```txt theme={null}
    fastapi==0.104.1
    uvicorn==0.24.0
    pydantic==2.5.0
    httpx==0.25.1
    python-dotenv==1.0.0
    ```

    Install:

    ```bash theme={null}
    pip install -r requirements.txt
    ```
  </Step>

  <Step title="Create MCP Server">
    **server.py**:

    ```python theme={null}
    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)
    ```
  </Step>

  <Step title="Run the Server">
    ```bash theme={null}
    python server.py
    ```

    Server will start on: `http://localhost:8766`
  </Step>

  <Step title="Test the Server">
    ```bash theme={null}
    # 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"}
        }
      }'
    ```
  </Step>
</Steps>

***

## Quick Start: Node.js MCP Server

### Project Setup

<Steps>
  <Step title="Initialize Project">
    ```bash theme={null}
    mkdir my-mcp-server-js
    cd my-mcp-server-js
    npm init -y
    npm install express body-parser axios dotenv
    ```
  </Step>

  <Step title="Create Server">
    **server.js**:

    ```javascript theme={null}
    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}`);
    });
    ```
  </Step>
</Steps>

***

## Adding Authentication

### API Key Authentication

<Tabs>
  <Tab title="Python (FastAPI)">
    ```python theme={null}
    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
    ```
  </Tab>

  <Tab title="Node.js (Express)">
    ```javascript theme={null}
    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
    });
    ```
  </Tab>
</Tabs>

### OAuth 2.0 Authentication

For production systems, implement OAuth 2.0:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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**:

```dockerfile theme={null}
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**:

```bash theme={null}
docker build -t my-mcp-server .
docker run -d -p 8766:8766 --name mcp-server my-mcp-server
```

### Production Considerations

<Check>
  **✅ 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)
</Check>

***

## 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**:

1. Go to **Team > MCP**
2. Click **"Connect"**
3. 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

1. Admin > Agents > \[Your Agent]
2. MCP Platforms > Check your custom platform
3. 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](https://github.com/modelcontextprotocol/servers) 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](https://github.com/modelcontextprotocol/servers/tree/main/src/postgres) 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

<Check>
  **✅ 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
</Check>

<Warning>
  **❌ 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
</Warning>

***

## What's Next?

<CardGroup cols={2}>
  <Card title="MCP Protocol Spec" icon="book" href="https://modelcontextprotocol.io/">
    Deep dive into MCP protocol details
  </Card>

  <Card title="MCP Overview" icon="list" href="/guides/mcp/overview">
    See how other platforms implement MCP
  </Card>

  <Card title="GitHub MCP" icon="github" href="/guides/mcp/github">
    Connect to GitHub repositories
  </Card>

  <Card title="Community MCPs" icon="users" href="https://github.com/topics/mcp-server">
    Browse open-source MCP servers
  </Card>
</CardGroup>

***

## Additional Resources

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

<Note>
  **Need Help?** Check the [MCP Protocol Specification](https://modelcontextprotocol.io/) and [MCP Server Examples](https://github.com/modelcontextprotocol/servers) for implementation guidance and best practices.
</Note>