Custom MCP Server - Awesome AI Apps

Overview

This example demonstrates how to build a custom MCP (Model Context Protocol) server using FastMCP and integrate it with AI agents. The example implements an email service that can configure SMTP settings and send emails through MCP tools.

Features

Custom MCP Server - Built with FastMCP framework
Tool Definition - Expose functions as MCP tools
Client Integration - Connect agents to custom server
Stdio Communication - Standard input/output protocol
Type Safety - Proper parameter typing and validation
Error Handling - Robust error management

Architecture

Prerequisites

Python 3.8+
FastMCP library
OpenAI Agents SDK
SMTP credentials (Gmail App Password)
Nebius API key

Installation

Clone Repository

cd ~/workspace/source/mcp_ai_agents/custom_mcp_server

Install Dependencies

pip install fastmcp openai agents dotenv

Configure Environment

Create a .env file:

NEBIUS_API_KEY=your_nebius_api_key
GOOGLE_PASSKEY=your_gmail_app_password

Building the MCP Server

Server Implementation

Create mcp-server.py:

from typing import Optional, Dict, Any
from mcp.server.fastmcp import FastMCP
import smtplib
from email.message import EmailMessage

# Initialize FastMCP server
mcp = FastMCP("email")

# Global variables for email configuration
SENDER_NAME: Optional[str] = None
SENDER_EMAIL: Optional[str] = None
SENDER_PASSKEY: Optional[str] = None

@mcp.tool()
def configure_email(
    sender_name: str,
    sender_email: str,
    sender_passkey: str
) -> Dict[str, Any]:
    """Configure email sender details.
    
    Args:
        sender_name: Name of the email sender
        sender_email: Email address of the sender
        sender_passkey: App password or passkey for email authentication
    """
    global SENDER_NAME, SENDER_EMAIL, SENDER_PASSKEY
    SENDER_NAME = sender_name
    SENDER_EMAIL = sender_email
    SENDER_PASSKEY = sender_passkey
    
    return {
        "success": True,
        "message": "Email configuration updated successfully"
    }

@mcp.tool()
def send_email(
    receiver_email: str,
    subject: str,
    body: str
) -> Dict[str, Any]:
    """Send an email to specified recipient.
    
    Args:
        receiver_email: Email address of the recipient
        subject: Subject line of the email
        body: Main content/body of the email
    
    Returns:
        Dictionary containing success status and message
    """
    if not all([SENDER_NAME, SENDER_EMAIL, SENDER_PASSKEY]):
        return {
            "success": False,
            "message": "Email sender not configured. Use configure_email first."
        }

    try:
        msg = EmailMessage()
        msg["Subject"] = subject
        msg["From"] = f"{SENDER_NAME} <{SENDER_EMAIL}>"
        msg["To"] = receiver_email
        msg.set_content(body)

        with smtplib.SMTP_SSL("smtp.gmail.com", 465) as smtp:
            smtp.login(SENDER_EMAIL, SENDER_PASSKEY)
            smtp.send_message(msg)
            
        return {
            "success": True,
            "message": "Email sent successfully"
        }
    except Exception as e:
        return {
            "success": False,
            "message": f"Error sending email: {str(e)}"
        }

if __name__ == "__main__":
    mcp.run(transport='stdio')

Key Components

FastMCP Initialization

from mcp.server.fastmcp import FastMCP

# Initialize with server name
mcp = FastMCP("email")

Tool Definition

@mcp.tool()
def configure_email(
    sender_name: str,
    sender_email: str,
    sender_passkey: str
) -> Dict[str, Any]:
    """Docstring becomes tool description."""
    # Implementation
    return {"success": True, "message": "..."}

The @mcp.tool() decorator exposes functions as MCP tools. The function’s docstring and type hints are used for tool metadata.

State Management

# Global state for configuration
SENDER_NAME: Optional[str] = None
SENDER_EMAIL: Optional[str] = None
SENDER_PASSKEY: Optional[str] = None

@mcp.tool()
def configure_email(...):
    global SENDER_NAME, SENDER_EMAIL, SENDER_PASSKEY
    # Update state

MCP servers can maintain state between tool calls.

Stdio Transport

if __name__ == "__main__":
    mcp.run(transport='stdio')

The server communicates via standard input/output, making it easy to spawn as a subprocess.

Building the MCP Client

Client Implementation

Create mcp-client.py:

import asyncio
import os
from agents import (
    Agent,
    OpenAIChatCompletionsModel,
    Runner,
    set_tracing_disabled,
)
from agents.mcp import MCPServerStdio
from openai import AsyncOpenAI
from dotenv import load_dotenv

load_dotenv()

# Constants
EMAIL_MCP_PATH = "/path/to/custom_mcp_server"
UV_PATH = "/path/to/uv"  # or use system python
MODEL_NAME = "meta-llama/Llama-3.3-70B-Instruct"
API_BASE_URL = "https://api.tokenfactory.nebius.com/v1"
PASSKEY = os.environ["GOOGLE_PASSKEY"]

async def setup_email_agent(mcp_server: MCPServerStdio) -> Agent:
    """Create and configure the Email agent."""
    return Agent(
        name="Email Assistant",
        instructions="""You are an email assistant that helps send emails. 
        First configure the email settings if not done, then help send 
        emails accurately.""",
        mcp_servers=[mcp_server],
        model=OpenAIChatCompletionsModel(
            model=MODEL_NAME,
            openai_client=AsyncOpenAI(
                base_url=API_BASE_URL,
                api_key=os.environ["NEBIUS_API_KEY"]
            )
        )
    )

async def main():
    try:
        async with MCPServerStdio(
            cache_tools_list=True,
            params={
                "command": UV_PATH,
                "args": [
                    "--directory",
                    EMAIL_MCP_PATH,
                    "run",
                    "mcp-server.py"
                ]
            }
        ) as mcp_server:
            email_agent = await setup_email_agent(mcp_server)
            
            # Example usage
            message = f"""Configure email with sender name 'AI Agent', 
            email '[email protected]', and passkey '{PASSKEY}' 
            then send an email to '[email protected]' with subject 
            'Test Email' and body 'Hello from Email MCP!'"""
            
            try:
                result = await Runner.run(
                    starting_agent=email_agent, 
                    input=message
                )
                print("\nEmail Result:")
                print(result.final_output)
            except Exception as e:
                print(f"\nError with email operation: {e}")
                
    except Exception as e:
        print(f"\nError initializing Email MCP server: {e}")

if __name__ == "__main__":
    set_tracing_disabled(disabled=True)
    asyncio.run(main())

Client Components

MCPServerStdio Configuration

from agents.mcp import MCPServerStdio

async with MCPServerStdio(
    cache_tools_list=True,  # Cache available tools
    params={
        "command": "uv",  # or "python"
        "args": [
            "--directory",
            "/path/to/server",
            "run",
            "mcp-server.py"
        ]
    }
) as mcp_server:
    # Use mcp_server with agent

Agent Configuration

from agents import Agent, OpenAIChatCompletionsModel

agent = Agent(
    name="Email Assistant",
    instructions="System prompt here",
    mcp_servers=[mcp_server],  # Connect to MCP server
    model=OpenAIChatCompletionsModel(
        model="meta-llama/Llama-3.3-70B-Instruct",
        openai_client=client
    )
)

Running the Agent

from agents import Runner

result = await Runner.run(
    starting_agent=agent,
    input="Your message here"
)

print(result.final_output)

Usage

Running the Server Standalone

python mcp-server.py

The server will wait for JSON-RPC messages on stdin.

Running with Client

python mcp-client.py

The client will:

Start the MCP server as subprocess
Create an agent connected to the server
Execute the email workflow
Display results

Advanced Patterns

Multiple Tools

Add multiple tools to your server:

@mcp.tool()
def send_email(...):
    pass

@mcp.tool()
def send_bulk_email(recipients: list[str], ...):
    pass

@mcp.tool()
def get_email_status(email_id: str):
    pass

@mcp.tool()
def list_sent_emails():
    pass

Resources

Expose data resources:

@mcp.resource("email://config")
def get_email_config():
    return {
        "sender_name": SENDER_NAME,
        "sender_email": SENDER_EMAIL,
        "configured": all([SENDER_NAME, SENDER_EMAIL, SENDER_PASSKEY])
    }

Prompts

Provide reusable prompts:

@mcp.prompt()
def compose_professional_email(recipient: str, purpose: str):
    return f"""Compose a professional email to {recipient} 
    for the purpose of {purpose}. Use appropriate greeting and closing."""

Context Management

Use FastMCP context for dependency injection:

from fastmcp import Context

@mcp.tool()
async def send_email_with_context(
    ctx: Context,
    receiver_email: str,
    subject: str,
    body: str
) -> Dict[str, Any]:
    # Access context data
    logger = ctx.get("logger")
    logger.info(f"Sending email to {receiver_email}")
    # ...

Error Handling

Robust error handling:

@mcp.tool()
def send_email(
    receiver_email: str,
    subject: str,
    body: str
) -> Dict[str, Any]:
    # Validation
    if not receiver_email or "@" not in receiver_email:
        return {
            "success": False,
            "message": "Invalid email address"
        }
    
    # Check configuration
    if not all([SENDER_NAME, SENDER_EMAIL, SENDER_PASSKEY]):
        return {
            "success": False,
            "message": "Email not configured. Call configure_email first."
        }
    
    # Try sending
    try:
        # ... send email ...
        return {"success": True, "message": "Email sent"}
    except smtplib.SMTPAuthenticationError:
        return {"success": False, "message": "Authentication failed"}
    except Exception as e:
        return {"success": False, "message": f"Error: {str(e)}"}

Testing

Unit Testing Tools

import pytest
from mcp_server import configure_email, send_email

def test_configure_email():
    result = configure_email(
        sender_name="Test",
        sender_email="[email protected]",
        sender_passkey="password"
    )
    assert result["success"] is True

def test_send_email_without_config():
    result = send_email(
        receiver_email="[email protected]",
        subject="Test",
        body="Test body"
    )
    assert result["success"] is False
    assert "not configured" in result["message"]

Integration Testing

import asyncio
from agents.mcp import MCPServerStdio
from agents import Agent, Runner

async def test_mcp_integration():
    async with MCPServerStdio(
        params={"command": "python", "args": ["mcp-server.py"]}
    ) as server:
        agent = Agent(
            name="Test Agent",
            mcp_servers=[server],
            model=...
        )
        
        result = await Runner.run(
            starting_agent=agent,
            input="Send a test email"
        )
        
        assert "success" in result.final_output.lower()

if __name__ == "__main__":
    asyncio.run(test_mcp_integration())

Deployment

Local Development

# Run server directly
python mcp-server.py

# Run with uv
uv run mcp-server.py

Production

# Use uv for isolated environment
uvx --from . run mcp-server

# Or with Docker
docker build -t custom-mcp-server .
docker run -i custom-mcp-server

Docker Example

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY mcp-server.py .

CMD ["python", "mcp-server.py"]

Best Practices

Type Safety

Use type hints for all parameters
Return consistent data structures
Validate input before processing
Document return value schemas

Error Handling

Return errors as data, not exceptions
Provide clear error messages
Include error codes when appropriate
Log errors for debugging

Documentation

Write clear docstrings for tools
Document required vs optional params
Provide usage examples
Explain side effects

State Management

Keep state minimal and clear
Use configuration tools for setup
Provide state inspection tools
Handle concurrent access

Troubleshooting

Server Not Starting

Check the command and args:

params={
    "command": "python",  # or full path
    "args": ["mcp-server.py"]  # ensure file exists
}

Tools Not Discovered

Ensure:

Server is running with transport='stdio'
Tools are decorated with @mcp.tool()
Client has cache_tools_list=True

Agent Can't Call Tools

Check:

mcp_servers=[mcp_server] in Agent config
Server subprocess is still running
No stdio communication errors

Source Code

View the complete implementation at: ~/workspace/source/mcp_ai_agents/custom_mcp_server/

FastMCP

FastMCP framework documentation

OpenAI Agents SDK

Database MCP Agent

GibsonAI database management

Couchbase MCP Server

Couchbase MCP integration

Starter Agents

Simple Agents

MCP Agents

Memory Agents

RAG Applications

Advanced Agents

​Overview

​Features

​Architecture

​Prerequisites

​Installation

​Building the MCP Server

​Server Implementation

​Key Components

​Building the MCP Client

​Client Implementation

​Client Components

​Usage

​Running the Server Standalone

​Running with Client

​Advanced Patterns

​Multiple Tools

​Resources

​Prompts

​Context Management

​Error Handling

​Testing

​Unit Testing Tools

​Integration Testing

​Deployment

​Local Development

​Production

​Docker Example

​Best Practices

Type Safety

Error Handling

Documentation

State Management

​Troubleshooting

​Source Code

FastMCP

OpenAI Agents SDK

​Related Examples

Database MCP Agent

Couchbase MCP Server

Build docs developers (and LLMs) love

Overview

Features

Architecture

Prerequisites

Installation

Building the MCP Server

Server Implementation

Key Components

Building the MCP Client

Client Implementation

Client Components

Usage

Running the Server Standalone

Running with Client

Advanced Patterns

Multiple Tools

Resources

Prompts

Context Management

Error Handling

Testing

Unit Testing Tools

Integration Testing

Deployment

Local Development

Production

Docker Example

Best Practices

Troubleshooting

Source Code

Related Examples