cnks 0.1.0

CNKI Search tool with Chrome browser integration

MCP Python SDK

Python implementation of the Model Context Protocol (MCP)

Table of Contents

• MCP Python SDK
  • Overview
  • Installation
    • Adding MCP to your python project
    • Running the standalone MCP development tools
  • Quickstart
  • What is MCP?
  • Core Concepts
    • Server
    • Resources
    • Tools
    • Prompts
    • Images
    • Context
  • Running Your Server
    • Development Mode
    • Claude Desktop Integration
    • Direct Execution
    • Mounting to an Existing ASGI Server
  • Examples
    • Echo Server
    • SQLite Explorer
  • Advanced Usage
    • Low-Level Server
    • Writing MCP Clients
    • MCP Primitives
    • Server Capabilities
  • Documentation
  • Contributing
  • License

Overview

The Model Context Protocol allows applications to provide context for LLMs in a standardized way, separating the concerns of providing context from the actual LLM interaction. This Python SDK implements the full MCP specification, making it easy to:

• Build MCP clients that can connect to any MCP server
• Create MCP servers that expose resources, prompts and tools
• Use standard transports like stdio and SSE
• Handle all MCP protocol messages and lifecycle events

Installation

Adding MCP to your python project

We recommend using uv to manage your Python projects.

If you haven't created a uv-managed project yet, create one:

uv init mcp-server-demo
cd mcp-server-demo

Then add MCP to your project dependencies:

uv add "mcp[cli]"

Alternatively, for projects using pip for dependencies:

pip install "mcp[cli]"

Running the standalone MCP development tools

To run the mcp command with uv:

uv run mcp

Quickstart

Let's create a simple MCP server that exposes a calculator tool and some data:

# 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


# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"

You can install this server in Claude Desktop and interact with it right away by running:

mcp install server.py

Alternatively, you can test it with the MCP Inspector:

mcp dev server.py

What is MCP?

The Model Context Protocol (MCP) lets you build servers that expose data and functionality to LLM applications in a secure, standardized way. Think of it like a web API, but specifically designed for LLM interactions. MCP servers can:

• Expose data through Resources (think of these sort of like GET endpoints; they are used to load information into the LLM's context)
• Provide functionality through Tools (sort of like POST endpoints; they are used to execute code or otherwise produce a side effect)
• Define interaction patterns through Prompts (reusable templates for LLM interactions)
• And more!

Core Concepts

Server

The FastMCP server is your core interface to the MCP protocol. It handles connection management, protocol compliance, and message routing:

# Add lifespan support for startup/shutdown with strong typing
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass

from fake_database import Database  # Replace with your actual DB type

from mcp.server.fastmcp import Context, FastMCP

# Create a named server
mcp = FastMCP("My App")

# Specify dependencies for deployment and development
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])


@dataclass
class AppContext:
    db: Database


@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manage application lifecycle with type-safe context"""
    # Initialize on startup
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        # Cleanup on shutdown
        await db.disconnect()


# Pass lifespan to server
mcp = FastMCP("My App", lifespan=app_lifespan)


# Access type-safe lifespan context in tools
@mcp.tool()
def query_db(ctx: Context) -> str:
    """Tool that uses initialized resources"""
    db = ctx.request_context.lifespan_context.db
    return db.query()

Resources

Resources are how you expose data to LLMs. They're similar to GET endpoints in a REST API - they provide data but shouldn't perform significant computation or have side effects:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")


@mcp.resource("config://app")
def get_config() -> str:
    """Static configuration data"""
    return "App configuration here"


@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """Dynamic user data"""
    return f"Profile data for user {user_id}"

Tools

Tools let LLMs take actions through your server. Unlike resources, tools are expected to perform computation and have side effects:

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")


@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
    """Calculate BMI given weight in kg and height in meters"""
    return weight_kg / (height_m**2)


@mcp.tool()
async def fetch_weather(city: str) -> str:
    """Fetch current weather for a city"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.weather.com/{city}")
        return response.text

Prompts

Prompts are reusable templates that help LLMs interact with your server effectively:

from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base

mcp = FastMCP("My App")


@mcp.prompt()
def review_code(code: str) -> str:
    return f"Please review this code:\n\n{code}"


@mcp.prompt()
def debug_error(error: str) -> list[base.Message]:
    return [
        base.UserMessage("I'm seeing this error:"),
        base.UserMessage(error),
        base.AssistantMessage("I'll help debug that. What have you tried so far?"),
    ]

Images

FastMCP provides an Image class that automatically handles image data:

from mcp.server.fastmcp import FastMCP, Image
from PIL import Image as PILImage

mcp = FastMCP("My App")


@mcp.tool()
def create_thumbnail(image_path: str) -> Image:
    """Create a thumbnail from an image"""
    img = PILImage.open(image_path)
    img.thumbnail((100, 100))
    return Image(data=img.tobytes(), format="png")

Context

The Context object gives your tools and resources access to MCP capabilities:

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("My App")


@mcp.tool()
async def long_task(files: list[str], ctx: Context) -> str:
    """Process multiple files with progress tracking"""
    for i, file in enumerate(files):
        ctx.info(f"Processing {file}")
        await ctx.report_progress(i, len(files))
        data, mime_type = await ctx.read_resource(f"file://{file}")
    return "Processing complete"

Running Your Server

Development Mode

The fastest way to test and debug your server is with the MCP Inspector:

mcp dev server.py

# Add dependencies
mcp dev server.py --with pandas --with numpy

# Mount local code
mcp dev server.py --with-editable .

Claude Desktop Integration

Once your server is ready, install it in Claude Desktop:

mcp install server.py

# Custom name
mcp install server.py --name "My Analytics Server"

# Environment variables
mcp install server.py -v API_KEY=abc123 -v DB_URL=postgres://...
mcp install server.py -f .env

Direct Execution

For advanced scenarios like custom deployments:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")

if __name__ == "__main__":
    mcp.run()

Run it with:

python server.py
# or
mcp run server.py

Mounting to an Existing ASGI Server

You can mount the SSE server to an existing ASGI server using the sse_app method. This allows you to integrate the SSE server with other ASGI applications.

from starlette.applications import Starlette
from starlette.routing import Mount, Host
from mcp.server.fastmcp import FastMCP


mcp = FastMCP("My App")

# Mount the SSE server to the existing ASGI server
app = Starlette(
    routes=[
        Mount('/', app=mcp.sse_app()),
    ]
)

# or dynamically mount as host
app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))

For more information on mounting applications in Starlette, see the Starlette documentation.

Examples

Echo Server

A simple server demonstrating resources, tools, and prompts:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Echo")


@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
    """Echo a message as a resource"""
    return f"Resource echo: {message}"


@mcp.tool()
def echo_tool(message: str) -> str:
    """Echo a message as a tool"""
    return f"Tool echo: {message}"


@mcp.prompt()
def echo_prompt(message: str) -> str:
    """Create an echo prompt"""
    return f"Please process this message: {message}"

SQLite Explorer

A more complex example showing database integration:

import sqlite3

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("SQLite Explorer")


@mcp.resource("schema://main")
def get_schema() -> str:
    """Provide the database schema as a resource"""
    conn = sqlite3.connect("database.db")
    schema = conn.execute("SELECT sql FROM sqlite_master WHERE type='table'").fetchall()
    return "\n".join(sql[0] for sql in schema if sql[0])


@mcp.tool()
def query_data(sql: str) -> str:
    """Execute SQL queries safely"""
    conn = sqlite3.connect("database.db")
    try:
        result = conn.execute(sql).fetchall()
        return "\n".join(str(row) for row in result)
    except Exception as e:
        return f"Error: {str(e)}"

Advanced Usage

Low-Level Server

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:

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

from fake_database import Database  # Replace with your actual DB type

from mcp.server import Server


@asynccontextmanager
async def server_lifespan(server: Server) -> AsyncIterator[dict]:
    """Manage server startup and shutdown lifecycle."""
    # Initialize resources on startup
    db = await Database.connect()
    try:
        yield {"db": db}
    finally:
        # Clean up on shutdown
        await db.disconnect()


# Pass lifespan to server
server = Server("example-server", lifespan=server_lifespan)


# Access lifespan context in handlers
@server.call_tool()
async def query_db(name: str, arguments: dict) -> list:
    ctx = server.request_context
    db = ctx.lifespan_context["db"]
    return await db.query(arguments["query"])

The lifespan API provides:

• A way to initialize resources when the server starts and clean them up when it stops
• Access to initialized resources through the request context in handlers
• Type-safe context passing between lifespan and request handlers

import mcp.server.stdio
import mcp.types as types
from mcp.server.lowlevel import NotificationOptions, Server
from mcp.server.models import InitializationOptions

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


@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
    return [
        types.Prompt(
            name="example-prompt",
            description="An example prompt template",
            arguments=[
                types.PromptArgument(
                    name="arg1", description="Example argument", required=True
                )
            ],
        )
    ]


@server.get_prompt()
async def handle_get_prompt(
    name: str, arguments: dict[str, str] | None
) -> types.GetPromptResult:
    if name != "example-prompt":
        raise ValueError(f"Unknown prompt: {name}")

    return types.GetPromptResult(
        description="Example prompt",
        messages=[
            types.PromptMessage(
                role="user",
                content=types.TextContent(type="text", text="Example prompt text"),
            )
        ],
    )


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())

Writing MCP Clients

The SDK provides a high-level client interface for connecting to MCP servers:

from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client

# Create server parameters for stdio connection
server_params = StdioServerParameters(
    command="python",  # Executable
    args=["example_server.py"],  # Optional command line arguments
    env=None,  # Optional environment variables
)


# Optional: create a sampling callback
async def handle_sampling_message(
    message: types.CreateMessageRequestParams,
) -> types.CreateMessageResult:
    return types.CreateMessageResult(
        role="assistant",
        content=types.TextContent(
            type="text",
            text="Hello, world! from model",
        ),
        model="gpt-3.5-turbo",
        stopReason="endTurn",
    )


async def run():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(
            read, write, sampling_callback=handle_sampling_message
        ) as session:
            # Initialize the connection
            await session.initialize()

            # List available prompts
            prompts = await session.list_prompts()

            # Get a prompt
            prompt = await session.get_prompt(
                "example-prompt", arguments={"arg1": "value"}
            )

            # List available resources
            resources = await session.list_resources()

            # List available tools
            tools = await session.list_tools()

            # Read a resource
            content, mime_type = await session.read_resource("file://some/path")

            # Call a tool
            result = await session.call_tool("tool-name", arguments={"arg1": "value"})


if __name__ == "__main__":
    import asyncio

    asyncio.run(run())

MCP Primitives

The MCP protocol defines three core primitives that servers can implement:

Primitive	Control	Description	Example Use
Prompts	User-controlled	Interactive templates invoked by user choice	Slash commands, menu options
Resources	Application-controlled	Contextual data managed by the client application	File contents, API responses
Tools	Model-controlled	Functions exposed to the LLM to take actions	API calls, data updates

Server Capabilities

MCP servers declare capabilities during initialization:

Capability	Feature Flag	Description
prompts	listChanged	Prompt template management
resources	subscribe listChanged	Resource exposure and updates
tools	listChanged	Tool discovery and execution
logging	-	Server logging configuration
completion	-	Argument completion suggestions

Documentation

• Model Context Protocol documentation
• Model Context Protocol specification
• Officially supported servers

Contributing

We are passionate about supporting contributors of all levels of experience and would love to see you get involved in the project. See the contributing guide to get started.

License

This project is licensed under the MIT License - see the LICENSE file for details.

知网搜索MCP服务器

该MCP服务器可以帮助您打开Chrome浏览器访问中国知网，并提供搜索功能。

功能

• 打开Chrome浏览器访问知网
• 提供搜索指定关键词的工具
• 包含常见关键词搜索的提示
• 允许用户保存笔记

安装

确保您已安装Python 3.13或更高版本，然后执行以下命令安装：

uv add "mcp[cli]"

安装Playwright（推荐）

为了实现自动填写搜索框和点击搜索按钮的功能，建议安装Playwright：

# 安装playwright库
uv add playwright

# 安装playwright浏览器
playwright install

如果不安装Playwright，搜索功能将仅打开知网页面而不会自动输入关键词。

开发模式运行

使用MCP Inspector测试和调试服务器：

mcp dev src/cnks/server.py

安装到Claude Desktop

准备好后，将其安装到Claude Desktop：

mcp install src/cnks/server.py --name "知网搜索助手"

在Cursor中连接

要在Cursor中成功连接到服务器，请按照以下步骤操作：

1. 确保使用Python 3.13或更高版本
2. 正确安装了所有依赖：uv add "mcp[cli]"
3. 在终端中直接启动服务器：

   python -m cnks
   

4. 在Cursor中使用MCP连接功能，选择"Connect to Server"

如果出现连接问题，可以尝试：

• 检查防火墙设置
• 确保没有其他程序占用相同端口
• 重启Cursor和服务器
• 确保使用最新版本的MCP

使用方法

资源

服务器提供以下资源：

• webpage://current - 当前打开的网页内容
• webpage://cnki/search - 中国知网搜索页面
• note://internal/{name} - 用户保存的笔记

工具

服务器提供以下工具：

• open-cnki - 打开中国知网搜索页面
• search-keywords - 在知网搜索关键词
• add-note - 添加笔记

提示

服务器提供以下提示模板：

• search-literature - 按主题搜索文献
• advanced-search - 高级文献搜索
• summarize-notes - 总结所有笔记

示例用法

打开知网

请打开中国知网搜索页面

搜索关键词

请在知网搜索"人工智能教育应用"相关论文

添加笔记

请添加一个笔记，名称为"AI教育论文"，内容为"人工智能在教育中的应用研究进展"

使用高级搜索

请使用高级搜索，查找标题包含"教育科技"、作者为"张三"的论文

故障排除

Chrome未找到

系统会自动在常见位置查找Chrome浏览器。如果遇到"未找到Chrome可执行文件"错误，可以通过设置环境变量来指定Chrome位置：

Windows

set CHROME_PATH="C:\你的Chrome路径\chrome.exe"

macOS/Linux

export CHROME_PATH="/path/to/chrome"

或者你也可以直接修改server.py文件中的Chrome检测代码，添加你的Chrome位置。

Playwright相关问题

如果在使用Playwright时遇到问题：

# 确保playwright库已安装
uv add playwright

# 安装所需浏览器
playwright install

# 如果上述命令无效，可能需要管理员权限

连接问题

如果在Cursor中连接服务器时遇到问题：

1. 尝试重启服务器和Cursor
2. 确认使用正确的连接方式（stdio或SSE）
3. 检查服务器日志输出是否有错误信息
4. 确保MCP依赖已正确安装：uv add "mcp[cli]" --upgrade

其他常见问题

• 中文字符显示问题：确保使用UTF-8编码
• 权限错误：在Windows上可能需要以管理员身份运行
• 网络访问受限：确保Chrome可以访问互联网

许可证

本项目使用MIT许可证 - 详情见LICENSE文件。