Uses Flask as the web framework to serve the MCP protocol endpoints, web interface, and Server-Sent Events streaming capabilities.
Supports deployment with Gunicorn for production environments with options for binding, port reuse, and hot reloading.
Provides integration with Azure OpenAI services for LLM completions, supporting both streaming and non-streaming responses with configurable models and deployments.
Includes metrics collection for monitoring request counts, response times, and error rates through a dedicated /metrics endpoint.
Implements Pydantic models for MCP protocol validation and type safety throughout the application.
Click on "Install Server".
Wait a few minutes for the server to deploy. Once ready, it will show a "Started" state.
In the chat, type
@followed by the MCP server name and your instructions, e.g., "@MCP Serversummarize the main features of this MCP server"
That's it! The server will respond to your query, and you can continue using it as needed.
Here is a step-by-step guide with screenshots.
MCP Server - Model Context Protocol Implementation
A comprehensive Python backend implementing the Model Context Protocol (MCP) with JSON-RPC 2.0, Azure OpenAI integration, and Server-Sent Events streaming capabilities.
Features
Complete MCP Protocol Support: JSON-RPC 2.0 compliant implementation
Azure OpenAI Integration: Seamless connection to Azure OpenAI services
Streaming Responses: Real-time streaming via Server-Sent Events (SSE)
Resource Management: File system resource discovery and access
Tool Execution: Extensible tool registry with validation
Authentication: JWT-based authentication system
Monitoring: Prometheus metrics collection
Web Interface: Built-in testing and management interface
Related MCP server: MCP Python SDK
Architecture
├── app/
│ ├── core/
│ │ ├── config.py # Configuration management
│ │ ├── errors.py # Custom exception classes
│ │ └── logging.py # Structured logging setup
│ ├── protocol/
│ │ ├── enums.py # MCP protocol enumerations
│ │ └── models.py # Pydantic models for MCP
│ ├── services/
│ │ ├── llm.py # Azure OpenAI service
│ │ ├── resources.py # Resource management
│ │ └── tools.py # Tool registry and execution
│ ├── transport/
│ │ └── http.py # HTTP transport layer
│ ├── auth.py # JWT authentication
│ └── metrics.py # Prometheus metrics
├── static/
│ └── app.js # Frontend JavaScript
├── templates/
│ └── index.html # Web interface
├── main.py # Application entry point
└── server.py # Flask app configurationInstallation
Clone the repository:
git clone <repository-url>
cd mcp-serverInstall dependencies:
pip install -r requirements.txtSet up environment variables:
# Required for Azure OpenAI
export OPENAI_API_KEY="your-azure-openai-api-key"
export AZURE_OPENAI_ENDPOINT="https://your-resource.openai.azure.com"
export AZURE_OPENAI_DEPLOYMENT="your-deployment-name"
export AZURE_OPENAI_API_VERSION="2024-08-01-preview"
# Optional configurations
export JWT_SECRET="your-jwt-secret"
export SESSION_SECRET="your-session-secret"Configuration
The server supports both Azure OpenAI and standard OpenAI configurations:
Azure OpenAI (Recommended)
USE_AZURE_OPENAI = True
AZURE_OPENAI_ENDPOINT = "https://your-resource.openai.azure.com"
AZURE_OPENAI_DEPLOYMENT = "gpt-4o"
AZURE_OPENAI_API_VERSION = "2024-08-01-preview"Standard OpenAI
USE_AZURE_OPENAI = False
OPENAI_MODEL = "gpt-4o"Running the Server
Development
python main.pyProduction
gunicorn --bind 0.0.0.0:5000 --reuse-port --reload main:appThe server will be available at http://localhost:5000
API Endpoints
MCP Protocol
POST /rpc- JSON-RPC 2.0 endpoint for MCP requestsGET /events- Server-Sent Events for streaming responses
Management
GET /- Web interface for testing and managementGET /health- Health check endpointGET /metrics- Prometheus metrics
Authentication
The server uses JWT-based authentication. Include the token in requests:
# HTTP Headers
Authorization: Bearer <token>
# Query Parameters (for SSE)
?token=<token>Default development token: devtoken
MCP Protocol Support
Capabilities
Resources: File system resource discovery and reading
Tools: Extensible tool execution with validation
Sampling: LLM completion requests (streaming and non-streaming)
Logging: Structured JSON logging
Example Requests
Initialize Connection
{
"jsonrpc": "2.0",
"id": "init",
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "test-client", "version": "1.0.0"}
}
}List Resources
{
"jsonrpc": "2.0",
"id": "resources",
"method": "resources/list",
"params": {}
}Execute Tool
{
"jsonrpc": "2.0",
"id": "tool",
"method": "tools/call",
"params": {
"name": "calculate",
"arguments": {"operation": "add", "a": 5, "b": 3}
}
}LLM Completion
{
"jsonrpc": "2.0",
"id": "completion",
"method": "sampling/createMessage",
"params": {
"messages": [{"content": {"type": "text", "text": "Hello, world!"}}],
"maxTokens": 100
}
}Extending the Server
Adding New Tools
from app.services.tools import mcp_tool
@mcp_tool("my_tool", {
"type": "object",
"properties": {
"param1": {"type": "string"},
"param2": {"type": "number"}
},
"required": ["param1"]
})
async def my_custom_tool(param1: str, param2: float = 0.0):
"""Custom tool implementation"""
return {"result": f"Processed {param1} with {param2}"}Custom Resource Handlers
from app.services.resources import ResourceService
class CustomResourceService(ResourceService):
async def list_resources(self, base_path: str = "."):
# Custom resource discovery logic
passMonitoring
The server includes comprehensive monitoring:
Prometheus Metrics: Request counts, response times, error rates
Structured Logging: JSON-formatted logs with correlation IDs
Health Checks: Application and dependency status
Security
Environment-based configuration (no hardcoded secrets)
JWT authentication with configurable secrets
Input validation on all endpoints
Rate limiting headers from Azure OpenAI
Development
Running Tests
# Test the API endpoints
curl -X POST http://localhost:5000/rpc \
-H "Authorization: Bearer devtoken" \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":"test","method":"initialize","params":{}}'
# Test streaming
curl -N "http://localhost:5000/events?token=devtoken&prompt=Hello&stream=true"Adding Dependencies
pip install <package-name>
pip freeze > requirements.txtTroubleshooting
Common Issues
Azure OpenAI Connection Errors
Verify
AZURE_OPENAI_ENDPOINTandAZURE_OPENAI_DEPLOYMENTCheck API key permissions
Ensure correct API version
Authentication Failures
Verify JWT token format
Check token expiration
Ensure correct secret configuration
Streaming Issues
Use query parameters for SSE authentication
Check network connectivity for long-running streams
Debug Logging
Enable debug logging by setting:
export DEBUG=trueLicense
This project is licensed under the MIT License.
Contributing
Fork the repository
Create a feature branch
Make your changes
Add tests for new functionality
Submit a pull request
Support
For issues and questions:
Check the troubleshooting section
Review the API documentation
Open an issue on GitHub