mcp_hello_server by akeredolukola - MCP Server

MCP Hello Server (Python)

Minimal MCP-style server implemented with FastAPI. It provides:

GET /health - simple health check
POST /mcp - accepts a small JSON payload (model + messages) and returns a simple assistant response

This project is intentionally small so you can run it locally and extend it to implement a real Model Context Protocol server.

Quick start (PowerShell):

Set-Location 'C:\Users\Cyphanet\mcp_hello_server_py'
python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
pip install -r requirements.txt
uvicorn app:app --host 127.0.0.1 --port 8000

Test the MCP endpoint with curl or http client:

curl -X POST "http://127.0.0.1:8000/mcp" -H "Content-Type: application/json" -d '{"model":"mcp-hello-1","messages":[{"role":"user","content":"Hello"}]}'

Run tests:

pytest -q

Extending the MCP Endpoint

The /mcp endpoint currently echoes back user messages. To implement real MCP behaviors, here are common patterns:

1. Streaming Responses

Use FastAPI's StreamingResponse to send chunks as they arrive:

from fastapi.responses import StreamingResponse
import asyncio

async def generate_streaming_response(messages):
    """Stream tokens as they are generated."""
    for token in ["Hello", " ", "from", " ", "MCP", "!"]:
        yield f"data: {token}\n"
        await asyncio.sleep(0.1)

@app.post("/mcp/stream")
async def mcp_stream(req: MCPRequest):
    return StreamingResponse(
        generate_streaming_response(req.messages),
        media_type="text/event-stream"
    )

2. Model Hooks

Create a hook system to inject custom logic before/after generation:

from typing import Callable

hooks: dict[str, list[Callable]] = {"pre_generate": [], "post_generate": []}

def register_hook(event: str, fn: Callable):
    """Register a hook function."""
    hooks[event].append(fn)

async def process_mcp_request(req: MCPRequest):
    # Pre-generation hooks
    for hook in hooks["pre_generate"]:
        await hook(req)
    
    # Generate response (your model call here)
    response = f"Echo: {req.messages}"
    
    # Post-generation hooks
    for hook in hooks["post_generate"]:
        await hook(response)
    
    return response

@app.post("/mcp")
async def mcp_endpoint(req: MCPRequest):
    content = await process_mcp_request(req)
    return {
        "model": req.model,
        "choices": [{"message": {"role": "assistant", "content": content}}]
    }

3. Integrating External Models

Replace the echo logic with calls to OpenAI, Claude, or your own model:

from openai import AsyncOpenAI

client = AsyncOpenAI(api_key="sk-...")

@app.post("/mcp")
async def mcp_endpoint(req: MCPRequest):
    response = await client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": m.role, "content": m.content} for m in req.messages]
    )
    return {
        "model": req.model,
        "choices": [{"message": {"role": "assistant", "content": response.choices[0].message.content}}]
    }

4. Request Validation & Error Handling

Extend Pydantic models and add exception handlers:

from fastapi import HTTPException
from pydantic import validator

class MCPRequest(BaseModel):
    model: str
    messages: List[Message]
    
    @validator("messages")
    def messages_not_empty(cls, v):
        if not v:
            raise ValueError("messages cannot be empty")
        return v

@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
    return JSONResponse(
        status_code=400,
        content={"detail": str(exc)}
    )

Start small with the echo logic, add one pattern at a time, and test with the provided pytest suite.