pythonmodel-context-protocol

Issue on call FastMCP Server using Postman


I have this simple MCP Server:

from fastmcp import FastMCP, Context

mcp = FastMCP("Demo 🚀")

@mcp.tool
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool
def hello(ctx: Context):
    ctx.info("in Hello!")
    return {"Respone": "Hello!"}


# Static resource
@mcp.resource("config://version")
def get_version(ctx: Context):
    ctx.info("Sono in get_version!")
    return "2.0.1"


if __name__ == "__main__":
    mcp.run(transport='streamable-http')

And I wish to test it using a REST call, but when I try:

curl --request POST \
  --url http://localhost:8000/mcp/tool/hello \
  --header 'accept: application/json, text/event-stream' \
  --header 'content-type: application/json' \
  --header 'x-session-id: my-test-session-123' \
  --data '{
  "jsonrpc": "2.0",
  "method": "hello",
  "id": 1
}

I got this error:

{
  "jsonrpc": "2.0",
  "id": "server-error",
  "error": {
    "code": -32600,
    "message": "Bad Request: Missing session ID"
  }
}

How can I resolve it?


Solution

  • You should understand how MCP life cycle works

    lifecycle

    MCP Protocol Lifecycle

    1. Initialize - Establishes session, exchanges capabilities

    2. Initialized notification - Tells server "I'm ready to use tools"

    3. Tool calls - Now you can actually call your functions

    The code correction

    from fastmcp import FastMCP, Context
    
    mcp = FastMCP("Demo 🚀")
    
    @mcp.tool
    def add(a: int, b: int) -> int:
        """Add two numbers"""
        return a + b
    
    @mcp.tool
    def hello(ctx: Context = None):
        """Say hello"""
        if ctx:
            ctx.info("in Hello!")
        return {"Response": "Hello!"}  #typo
    
    # Static resource
    @mcp.resource("config://version")
    def get_version(ctx: Context):
        ctx.info("Sono in get_version!")
        return "2.0.1"
    
    if __name__ == "__main__":
        mcp.run(transport='streamable-http')
    

    Powershell
    1. Initialize session

    
    $init = Invoke-WebRequest -Uri "http://localhost:8000/mcp" -Method POST -Headers @{'accept'='application/json, text/event-stream'; 'content-type'='application/json'} -Body '{"jsonrpc":"2.0","method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0.0"}},"id":1}'
    
    $sessionId = $init.Headers['mcp-session-id']
    

    Outcomes - sessionId

    2. Send initialized notification - pass sessionID here

    Invoke-RestMethod -Uri "http://localhost:8000/mcp" -Method POST -Headers @{'accept'='application/json, text/event-stream'; 'content-type'='application/json'; 'Mcp-Session-Id'=$sessionId} -Body '{"jsonrpc":"2.0","method":"notifications/initialized"}' | Out-Null
    

    3. call you function.

    $response = Invoke-RestMethod -Uri "http://localhost:8000/mcp" -Method POST -Headers @{'accept'='application/json, text/event-stream'; 'content-type'='application/json'; 'Mcp-Session-Id'=$sessionId} -Body '{"jsonrpc":"2.0","method":"tools/call","params":{"name":"add","arguments":{"a":10,"b":15}},"id":2}'
    

    4. Parse result

    $lines = $response -split "`n"
    $dataLine = $lines | Where-Object {$_ -like "data: *"}
    $jsonData = $dataLine.Substring(6)
    $result = $jsonData | ConvertFrom-Json
    Write-Host "result: $($result.result.structuredContent.result)"
    
    The output
    Result: 25
    

    Python version

    import requests
    import json
    
    def test_fastmcp_server():
       base_url = "http://localhost:8000/mcp"
       headers = {
           'accept': 'application/json, text/event-stream',
           'content-type': 'application/json'
       }
       
       print("check 1")
       
       init_payload = {
           "jsonrpc": "2.0",
           "method": "initialize",
           "params": {
               "protocolVersion": "2024-11-05",
               "capabilities": {},
               "clientInfo": {
                   "name": "python-client",
                   "version": "1.0.0"
               }
           },
           "id": 1
       }
       
       response = requests.post(base_url, headers=headers, json=init_payload)
       session_id = response.headers.get('mcp-session-id')
       print(f"Session ID: {session_id}")
       
       if not session_id:
           print("No session ID received")
           return
       
       headers['Mcp-Session-Id'] = session_id
       
       init_complete_payload = {
           "jsonrpc": "2.0",
           "method": "notifications/initialized"
       }
       
       requests.post(base_url, headers=headers, json=init_complete_payload)
       print("Initialization complete")
       
       add_payload = {
           "jsonrpc": "2.0",
           "method": "tools/call",
           "params": {
               "name": "add",
               "arguments": {
                   "a": 10,
                   "b": 15
               }
           },
           "id": 2
       }
       
       response = requests.post(base_url, headers=headers, json=add_payload)
       
       lines = response.text.split('\n')
       data_line = next((line for line in lines if line.startswith('data: ')), None)
       
       if data_line:
           json_data = data_line[6:]
           result = json.loads(json_data)
           answer = result['result']['structuredContent']['result']
           print(f"Add result: {answer}")
       else:
           print("No data found in response")
           print("Raw response:", response.text)
       
       hello_payload = {
           "jsonrpc": "2.0",
           "method": "tools/call",
           "params": {
               "name": "hello",
               "arguments": {}
           },
           "id": 3
       }
       
       response = requests.post(base_url, headers=headers, json=hello_payload)
       lines = response.text.split('\n')
       data_line = next((line for line in lines if line.startswith('data: ')), None)
       
       if data_line:
           json_data = data_line[6:]
           result = json.loads(json_data)
           hello_response = result['result']['content'][0]['text']
           print(f"Hello response: {hello_response}")
       
       print("Test complete")
    
    if __name__ == "__main__":
       try:
           test_fastmcp_server()
       except requests.exceptions.ConnectionError:
           print("Check MCP its not working on 8000")
       except Exception as e:
           print(f"Error: {e}")
    

    The output

    
    check 1
    Session ID: c15a125556ae4ea6bb12a5d62107e27a
    Initialization complete
    Add result: 25
    Hello response: {"Response":"Hello!"}
    Test complete