nacos-mcp-wrapper-python


Namenacos-mcp-wrapper-python JSON
Version 1.0.8 PyPI version JSON
download
home_pagehttps://github.com/nacos-group/nacos-mcp-wrapper-python
SummaryPython sdk support mcp server auto register to nacos
upload_time2025-07-29 03:38:08
maintainerNone
docs_urlNone
authornacos
requires_python>=3.10
licenseApache License 2.0
keywords
VCS
bugtrack_url
requirements psutil anyio mcp nacos-sdk-python pydantic pydantic-settings jsonref uvicorn nacos-maintainer-sdk-python
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # nacos-mcp-wrapper-python

[中文文档](./README_CN.md)  

## Overview
Nacos-mcp-wrapper-python is a sdk that helps you quickly register your Mcp Server to Nacos. Nacos is an easy-to-use platform designed for dynamic service discovery and configuration and service management. It helps you to build cloud native applications and microservices platform easily. By using Nacos to host your Mcp Server, it supports dynamic modifications of the descriptions for Mcp Server Tools and their corresponding parameters, assisting in the rapid evolution of your Mcp Server.

## Installation

### Environment
Starting from version 1.0.0 of nacos-mcp-wrapper-python, the Nacos server version must be greater than 3.0.1.
- python >=3.10
### Use pip
```bash
pip install nacos-mcp-wrapper-python
```

## Development
We can use official MCP Python SDK to quickly build a mcp server:
```python
# server.py
from mcp.server.fastmcp import FastMCP

# Create an MCP server
mcp = FastMCP("Demo")


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

mcp.run(transport="sse")
```
To quickly register your Mcp Server to Nacos, just replace the FashMCP with NacosMCP:

```python
# server.py
from nacos_mcp_wrapper.server.nacos_mcp import NacosMCP
from nacos_mcp_wrapper.server.nacos_settings import NacosSettings

# Create an MCP server
# mcp = FastMCP("Demo")
nacos_settings = NacosSettings()
nacos_settings.SERVER_ADDR = "<nacos_server_addr> e.g.127.0.0.1:8848"
mcp = NacosMCP("nacos-mcp-python", nacos_settings=nacos_settings,
               port=18001,
               instructions="This is a simple Nacos MCP server",
               version="1.0.0")


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

mcp.run(transport="sse")
```
After registering to Nacos, you can dynamically update the descriptions of Tools and the descriptions of parameters in the Mcp Server on Nacos without restarting your Mcp Server.

### Advanced Usage

When building an MCP server using the official MCP Python SDK, for more control, you can directly use the low-level server implementation, for more control, you can use the low-level server implementation directly. This gives you full access to the protocol and allows you to customize every aspect of your server, including lifecycle management through the lifespan API.
```python
import mcp.server.stdio
import mcp.types as types
import httpx
from mcp.server.lowlevel import NotificationOptions, Server
from mcp.server.models import InitializationOptions

# Create a server instance
server = Server("example-server")


async def fetch_website(
    url: str,
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    headers = {
        "User-Agent": "MCP Test Server (github.com/modelcontextprotocol/python-sdk)"
    }
    async with httpx.AsyncClient(follow_redirects=True, headers=headers) as client:
        response = await client.get(url)
        response.raise_for_status()
        return [types.TextContent(type="text", text=response.text)]
    
@server.call_tool()
async def fetch_tool(
    name: str, arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    if name != "fetch":
        raise ValueError(f"Unknown tool: {name}")
    if "url" not in arguments:
        raise ValueError("Missing required argument 'url'")
    return await fetch_website(arguments["url"])

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="fetch",
            description="Fetches a website and returns its content",
            inputSchema={
                "type": "object",
                "required": ["url"],
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "URL to fetch",
                    }
                },
            },
        )
    ]


async def run():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="example",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )


if __name__ == "__main__":
    import asyncio

    asyncio.run(run())
```

To quickly register your Mcp Server to Nacos, just replace the Server with NacosServer:

```python
import mcp.server.stdio
import mcp.types as types
import httpx
from mcp.server.lowlevel import NotificationOptions
from mcp.server.models import InitializationOptions
from nacos_mcp_wrapper.server.nacos_server import NacosServer
from nacos_mcp_wrapper.server.nacos_settings import NacosSettings

# Create a server instance
# server = Server("example-server")
nacos_settings = NacosSettings()
nacos_settings.SERVER_ADDR = "<nacos_server_addr> e.g.127.0.0.1:8848"
server = NacosServer("mcp-website-fetcher",nacos_settings=nacos_settings)

async def fetch_website(
    url: str,
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    headers = {
        "User-Agent": "MCP Test Server (github.com/modelcontextprotocol/python-sdk)"
    }
    async with httpx.AsyncClient(follow_redirects=True, headers=headers) as client:
        response = await client.get(url)
        response.raise_for_status()
        return [types.TextContent(type="text", text=response.text)]
    
@server.call_tool()
async def fetch_tool(
    name: str, arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    if name != "fetch":
        raise ValueError(f"Unknown tool: {name}")
    if "url" not in arguments:
        raise ValueError("Missing required argument 'url'")
    return await fetch_website(arguments["url"])

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="fetch",
            description="Fetches a website and returns its content",
            inputSchema={
                "type": "object",
                "required": ["url"],
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "URL to fetch",
                    }
                },
            },
        )
    ]


async def run():
    await server.register_to_nacos("stdio")
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="example",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )


if __name__ == "__main__":
    import asyncio

    asyncio.run(run())

```

For more examples, please refer to the content under the `example` directory.


            

Raw data

            {
    "_id": null,
    "home_page": "https://github.com/nacos-group/nacos-mcp-wrapper-python",
    "name": "nacos-mcp-wrapper-python",
    "maintainer": null,
    "docs_url": null,
    "requires_python": ">=3.10",
    "maintainer_email": null,
    "keywords": null,
    "author": "nacos",
    "author_email": null,
    "download_url": "https://files.pythonhosted.org/packages/9c/c5/01f710efaf3e5392ba60d0b950b7217a1486c2a98484d9cb3abcbf8adafa/nacos_mcp_wrapper_python-1.0.8.tar.gz",
    "platform": null,
    "description": "# nacos-mcp-wrapper-python\n\n[\u4e2d\u6587\u6587\u6863](./README_CN.md)  \n\n## Overview\nNacos-mcp-wrapper-python is a sdk that helps you quickly register your Mcp Server to Nacos. Nacos is an easy-to-use platform designed for dynamic service discovery and configuration and service management. It helps you to build cloud native applications and microservices platform easily. By using Nacos to host your Mcp Server, it supports dynamic modifications of the descriptions for Mcp Server Tools and their corresponding parameters, assisting in the rapid evolution of your Mcp Server.\n\n## Installation\n\n### Environment\nStarting from version 1.0.0 of nacos-mcp-wrapper-python, the Nacos server version must be greater than 3.0.1.\n- python >=3.10\n### Use pip\n```bash\npip install nacos-mcp-wrapper-python\n```\n\n## Development\nWe can use official MCP Python SDK to quickly build a mcp server:\n```python\n# server.py\nfrom mcp.server.fastmcp import FastMCP\n\n# Create an MCP server\nmcp = FastMCP(\"Demo\")\n\n\n# Add an addition tool\n@mcp.tool()\ndef add(a: int, b: int) -> int:\n    \"\"\"Add two numbers\"\"\"\n    return a + b\n\nmcp.run(transport=\"sse\")\n```\nTo quickly register your Mcp Server to Nacos, just replace the FashMCP with NacosMCP:\n\n```python\n# server.py\nfrom nacos_mcp_wrapper.server.nacos_mcp import NacosMCP\nfrom nacos_mcp_wrapper.server.nacos_settings import NacosSettings\n\n# Create an MCP server\n# mcp = FastMCP(\"Demo\")\nnacos_settings = NacosSettings()\nnacos_settings.SERVER_ADDR = \"<nacos_server_addr> e.g.127.0.0.1:8848\"\nmcp = NacosMCP(\"nacos-mcp-python\", nacos_settings=nacos_settings,\n               port=18001,\n               instructions=\"This is a simple Nacos MCP server\",\n               version=\"1.0.0\")\n\n\n# Add an addition tool\n@mcp.tool()\ndef add(a: int, b: int) -> int:\n    \"\"\"Add two numbers\"\"\"\n    return a + b\n\nmcp.run(transport=\"sse\")\n```\nAfter registering to Nacos, you can dynamically update the descriptions of Tools and the descriptions of parameters in the Mcp Server on Nacos without restarting your Mcp Server.\n\n### Advanced Usage\n\nWhen building an MCP server using the official MCP Python SDK, for more control, you can directly use the low-level server implementation, for more control, you can use the low-level server implementation directly. This gives you full access to the protocol and allows you to customize every aspect of your server, including lifecycle management through the lifespan API.\n```python\nimport mcp.server.stdio\nimport mcp.types as types\nimport httpx\nfrom mcp.server.lowlevel import NotificationOptions, Server\nfrom mcp.server.models import InitializationOptions\n\n# Create a server instance\nserver = Server(\"example-server\")\n\n\nasync def fetch_website(\n    url: str,\n) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:\n    headers = {\n        \"User-Agent\": \"MCP Test Server (github.com/modelcontextprotocol/python-sdk)\"\n    }\n    async with httpx.AsyncClient(follow_redirects=True, headers=headers) as client:\n        response = await client.get(url)\n        response.raise_for_status()\n        return [types.TextContent(type=\"text\", text=response.text)]\n    \n@server.call_tool()\nasync def fetch_tool(\n    name: str, arguments: dict\n) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:\n    if name != \"fetch\":\n        raise ValueError(f\"Unknown tool: {name}\")\n    if \"url\" not in arguments:\n        raise ValueError(\"Missing required argument 'url'\")\n    return await fetch_website(arguments[\"url\"])\n\n@server.list_tools()\nasync def list_tools() -> list[types.Tool]:\n    return [\n        types.Tool(\n            name=\"fetch\",\n            description=\"Fetches a website and returns its content\",\n            inputSchema={\n                \"type\": \"object\",\n                \"required\": [\"url\"],\n                \"properties\": {\n                    \"url\": {\n                        \"type\": \"string\",\n                        \"description\": \"URL to fetch\",\n                    }\n                },\n            },\n        )\n    ]\n\n\nasync def run():\n    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):\n        await server.run(\n            read_stream,\n            write_stream,\n            InitializationOptions(\n                server_name=\"example\",\n                server_version=\"0.1.0\",\n                capabilities=server.get_capabilities(\n                    notification_options=NotificationOptions(),\n                    experimental_capabilities={},\n                ),\n            ),\n        )\n\n\nif __name__ == \"__main__\":\n    import asyncio\n\n    asyncio.run(run())\n```\n\nTo quickly register your Mcp Server to Nacos, just replace the Server with NacosServer:\n\n```python\nimport mcp.server.stdio\nimport mcp.types as types\nimport httpx\nfrom mcp.server.lowlevel import NotificationOptions\nfrom mcp.server.models import InitializationOptions\nfrom nacos_mcp_wrapper.server.nacos_server import NacosServer\nfrom nacos_mcp_wrapper.server.nacos_settings import NacosSettings\n\n# Create a server instance\n# server = Server(\"example-server\")\nnacos_settings = NacosSettings()\nnacos_settings.SERVER_ADDR = \"<nacos_server_addr> e.g.127.0.0.1:8848\"\nserver = NacosServer(\"mcp-website-fetcher\",nacos_settings=nacos_settings)\n\nasync def fetch_website(\n    url: str,\n) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:\n    headers = {\n        \"User-Agent\": \"MCP Test Server (github.com/modelcontextprotocol/python-sdk)\"\n    }\n    async with httpx.AsyncClient(follow_redirects=True, headers=headers) as client:\n        response = await client.get(url)\n        response.raise_for_status()\n        return [types.TextContent(type=\"text\", text=response.text)]\n    \n@server.call_tool()\nasync def fetch_tool(\n    name: str, arguments: dict\n) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:\n    if name != \"fetch\":\n        raise ValueError(f\"Unknown tool: {name}\")\n    if \"url\" not in arguments:\n        raise ValueError(\"Missing required argument 'url'\")\n    return await fetch_website(arguments[\"url\"])\n\n@server.list_tools()\nasync def list_tools() -> list[types.Tool]:\n    return [\n        types.Tool(\n            name=\"fetch\",\n            description=\"Fetches a website and returns its content\",\n            inputSchema={\n                \"type\": \"object\",\n                \"required\": [\"url\"],\n                \"properties\": {\n                    \"url\": {\n                        \"type\": \"string\",\n                        \"description\": \"URL to fetch\",\n                    }\n                },\n            },\n        )\n    ]\n\n\nasync def run():\n    await server.register_to_nacos(\"stdio\")\n    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):\n        await server.run(\n            read_stream,\n            write_stream,\n            InitializationOptions(\n                server_name=\"example\",\n                server_version=\"0.1.0\",\n                capabilities=server.get_capabilities(\n                    notification_options=NotificationOptions(),\n                    experimental_capabilities={},\n                ),\n            ),\n        )\n\n\nif __name__ == \"__main__\":\n    import asyncio\n\n    asyncio.run(run())\n\n```\n\nFor more examples, please refer to the content under the `example` directory.\n\n",
    "bugtrack_url": null,
    "license": "Apache License 2.0",
    "summary": "Python sdk support mcp server auto register to nacos",
    "version": "1.0.8",
    "project_urls": {
        "Homepage": "https://github.com/nacos-group/nacos-mcp-wrapper-python"
    },
    "split_keywords": [],
    "urls": [
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "9e6b135011f85239f34419d62c53c41da91e3a6b8087750348d96e90938601e0",
                "md5": "5e4284fa1ef8a12e77252c4b208bb88e",
                "sha256": "681bfee2b2b18e79af9064e11272b2cb02d599b225b9e2ace386881e85f5aae1"
            },
            "downloads": -1,
            "filename": "nacos_mcp_wrapper_python-1.0.8-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "5e4284fa1ef8a12e77252c4b208bb88e",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.10",
            "size": 14577,
            "upload_time": "2025-07-29T03:38:04",
            "upload_time_iso_8601": "2025-07-29T03:38:04.308462Z",
            "url": "https://files.pythonhosted.org/packages/9e/6b/135011f85239f34419d62c53c41da91e3a6b8087750348d96e90938601e0/nacos_mcp_wrapper_python-1.0.8-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "9cc501f710efaf3e5392ba60d0b950b7217a1486c2a98484d9cb3abcbf8adafa",
                "md5": "f71d8800e1e239b3a20de0035a658158",
                "sha256": "f5accea3a1262ed6576f74d2282f7867bedaeb743444dd50025572841bc6e1b4"
            },
            "downloads": -1,
            "filename": "nacos_mcp_wrapper_python-1.0.8.tar.gz",
            "has_sig": false,
            "md5_digest": "f71d8800e1e239b3a20de0035a658158",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.10",
            "size": 15209,
            "upload_time": "2025-07-29T03:38:08",
            "upload_time_iso_8601": "2025-07-29T03:38:08.620772Z",
            "url": "https://files.pythonhosted.org/packages/9c/c5/01f710efaf3e5392ba60d0b950b7217a1486c2a98484d9cb3abcbf8adafa/nacos_mcp_wrapper_python-1.0.8.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2025-07-29 03:38:08",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "github_user": "nacos-group",
    "github_project": "nacos-mcp-wrapper-python",
    "travis_ci": false,
    "coveralls": false,
    "github_actions": false,
    "requirements": [
        {
            "name": "psutil",
            "specs": [
                [
                    "==",
                    "7.0.0"
                ]
            ]
        },
        {
            "name": "anyio",
            "specs": [
                [
                    "==",
                    "4.9.0"
                ]
            ]
        },
        {
            "name": "mcp",
            "specs": [
                [
                    "==",
                    "1.9.2"
                ]
            ]
        },
        {
            "name": "nacos-sdk-python",
            "specs": [
                [
                    ">=",
                    "2.0.8"
                ]
            ]
        },
        {
            "name": "pydantic",
            "specs": [
                [
                    "==",
                    "2.11.3"
                ]
            ]
        },
        {
            "name": "pydantic-settings",
            "specs": [
                [
                    "==",
                    "2.9.1"
                ]
            ]
        },
        {
            "name": "jsonref",
            "specs": [
                [
                    "==",
                    "1.1.0"
                ]
            ]
        },
        {
            "name": "uvicorn",
            "specs": [
                [
                    "==",
                    "0.34.2"
                ]
            ]
        },
        {
            "name": "nacos-maintainer-sdk-python",
            "specs": [
                [
                    ">=",
                    "0.1.2"
                ]
            ]
        }
    ],
    "lcname": "nacos-mcp-wrapper-python"
}
        
Elapsed time: 2.49992s