Master MCP Server and Client 01

Master MCP Server|Client

Why use MCP?

I won’t explain what MCP is. Recently, blogs and tweets related to MCP have become everywhere. I will only talk about the hardcore stuff here, which you don’t usually see.

In my opinion, the biggest difference between MCP and traditional tool calls is the dynamic pluggable feature:

The tools that can be called by a tool component are dynamic. When we add a new tool or delete a tool in a tool group, all clients that can connect to the current "tool group server" can sense it. This is very useful.

Of course, in addition to its versatility, the hierarchical architecture is clearer, splitting concepts such as resource, tools, and root.

In addition, whether a technology is worth learning or not, there is only one standard, which is "ecosystem" and "endorsement from big companies" . There is no doubt about this. In addition to OpenAI's public compatibility, Alibaba and Tencent have begun to fully support it. This is a major trend of tool call.

tool call From the primitive "slash-and-burn" era, we entered the "Bronze Age".

SSE Transport

The official example is based on the stdio mode "https://modelcontextprotocol.io/quickstart/server" , which is the original mode.

In this mode, your server and client need to be on the same machine. To start the client, go to https://modelcontextprotocol.io/quickstart/client :

uv run client.py path/to/server.py  # python server

The client code is as follows:

server_params = StdioServerParameters(
  command = "python" ,
  args=[server_script_path],
  env= None
)

stdio_transport =  await  self.exit_stack.enter_async_context(stdio_client(server_params))

99% of the tutorials on the Internet are of this type, but do you have a "worry" : if we need a lot of servers, all of them have to be on one machine, and if we want to make changes, all the machines have to be restarted. Isn't this going back to the original embarrassing situation?

MCP certainly thinks of what you think of. Therefore, in addition to the stdio ( "Standard Input/Output" ) mode, MCP also supports the sse ( "Server-Sent Events" ) mode, which is a network-based remote calling method.

def run (self, transport: Literal[ "stdio" ,  "sse" ] =  "stdio" ) 
....

The official explanation is:

❝

The SSE transport implements client-to-server communication by streaming from server to client via HTTP POST requests.

Use SSE when:

• Only server-to-client streaming is required
• Using restricted networks
• Implementing simple updates

❞

Because it is a network-based working method, the official has added additional "notes" for SSE :

❝

Security Warning: DNS Rebinding Attacks

If not properly protected, SSE transports can be vulnerable to DNS rebinding attacks. To prevent this:

1. Always validate the Origin header on incoming SSE connections to ensure they are coming from the expected origin
2. When running locally, avoid binding the server to all network interfaces (0.0.0.0) - only bind to localhost (127.0.0.1)
3. Implement proper authentication for all SSE connections

Without these protections, an attacker could use DNS rebinding to interact with the local MCP server from a remote website.

❞

OK, the basic preparations have been completed, and now we are ready to start the long-awaited demo session.

SSE Server

For building Server, the official SDK builds "mcp.server.fastmcp.FastMCP" for fast building. Of course, you can also use "mcp.server.Server" to build, the principle is the same, FastMCP encapsulates a layer, which is more convenient.

Because this is a sample demo, we will only build it in the simplest way here: build two classic tools: weather query and currency exchange rate conversion.

1 ⚙ Initialize FastMCP server

from  mcp.server.fastmcp  import  FastMCP

# The default host is 0.0.0.0 and the port is 8000
mcp = FastMCP( "sse_weather" , host= "localhost" , port= 9990 )

For more configurable items, see "mcp.server.fastmcp.server.Settings"  class.

2? Write a tool. This looks familiar, similar to the "langchain" method.

@mcp.tool()
async def get_weather (city: str)  -> str:  
    """Get weather information for a city.

    Args:
        city: Name of the city
    """

    return f" {city}  is sunny, enjoy it!" 

@mcp.tool()
def convert (amount: float, currency_from: str, currency_to: str)  -> float: 
    """use latest exchange rate to convert currency

    Args:
      amount: the amount of currency to convert
      currency_from: the currency to convert from
      currency_to: the currency to convert to
    """
    return  amount *  0.8

3 ? Run the server

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

Everything??, is there Cheng Yaojin's three boards?

I am using uv 「https://docs.astral.sh/uv」 as the Python package manager.

This tool is very convenient to use, and has a strong flavor of Rust package management (because it is developed in Rust). The latest version of pychaorm also supports uv as a package manager, which makes it very convenient to create projects.

Save the above code as sse_weather.py

So I started it using the following command:

 uv run sse_weather.py

If you see the following display, it is successful

INFO: Started server process [24973]
INFO: Waiting  for  application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://localhost:9990 (Press CTRL+C to quit)

SSE Client

1 Create a Client object

import  asyncio
from  contextlib  import  AsyncExitStack
from  typing  import  Optional

from  mcp  import  ClientSession, ListToolsResult
from  mcp.client.sse  import  sse_client
from  mcp.types  import  TextContent, ImageContent, EmbeddedResource, Tool, CallToolResult


class MCPClient : 
    def __init__ (self) : 
        #Initialize session and client objects
        self.session: Optional[ClientSession] =  None
        self.exit_stack = AsyncExitStack()

2? Add a method to connect to the server

    async def connect_to_server (self, server_url: str) :  
        """Connect to an MCP server

        Args:
            server_url: URL of the server
        """

        sc = sse_client(url=server_url)
        sse_transport =  await  self.exit_stack.enter_async_context(sc)
        sse, write = sse_transport
        cs = ClientSession(sse, write)
        self.session =  await  self.exit_stack.enter_async_context(cs)

        await  self.session.initialize()

3 ? Add tool_list method and tool_call method

    async def list_tools (self)  -> list[Tool]:  
        """List available tools"""
        response: ListToolsResult =  await  self.session.list_tools()
        print(response.model_dump())
        return  response.tools


    async def tool_call (self, tool_name: str, args: dict)  -> list[TextContent | ImageContent | EmbeddedResource]:  
        """Call a tool by name with arguments"""
        response: CallToolResult =  await  self.session.call_tool(tool_name, args)
        return  response.content

4 ♻️ Resource cleanup method

    async def cleanup (self) :  
        """Clean up resources"""
        await  self.exit_stack.aclose()

Then you can test it:

async def main () :  
    client = MCPClient()
    try :
        await  client.connect_to_server( "http://127.0.0.1:9990/sse" )
        await  client.list_tools()
        r =  await  client.tool_call( "get_weather" , { "city" :  "Beijing" })
        print(r)

    finally :
        await  client.cleanup()

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

Similarly, you can use uv to start

uv run sse_client.py

Take a look at the output

First, let’s take a look at the “ListToolsResult” object:

{
  "meta" :  "None" ,
"nextCursor" :  "None" ,
"tools" : [
    {
      "name" :  "get_weather" ,
      "description" :  "Get weather information for a city.\n\n Args:\n city: Name of the city\n " ,
      "inputSchema" : {
        "properties" : {
          "city" : {
            "title" :  "City" ,
            "type" :  "string"
          }
        },
        "required" : [
          "city"
        ],
        "title" :  "get_weatherArguments" ,
        "type" :  "object"
      }
    },
    {
      "name" :  "convert" ,
      "description" :  "use latest exchange rate to convert currency\n\n Args:\n amount: the amount of currency to convert\n currency_from: the currency to convert from\n currency_to: the currency to convert to\n " ,
      "inputSchema" : {
        "properties" : {
          "amount" : {
            "title" :  "Amount" ,
            "type" :  "number"
          },
          "currency_from" : {
            "title" :  "Currency From" ,
            "type" :  "string"
          },
          "currency_to" : {
            "title" :  "Currency To" ,
            "type" :  "string"
          }
        },
        "required" : [
          "amount" ,
          "currency_from" ,
          "currency_to"
        ],
        "title" :  "convertArguments" ,
        "type" :  "object"
      }
    }
  ]
}

This format should be familiar to you.

Then after calling the weather tool, the output on the server is:

[TextContent( type = 'text' , text= 'Beijing is sunny, enjoy it!' , annotations=None)]

At this point, the basics have been introduced. Later we will talk about how to combine it with the familiar LLM to implement tool call