IUHHUI

#!/usr/bin/env python3
"""
SSE (Server-Sent Events) MCP 服务测试脚本
用于验证 SSE 方式的 MCP 服务功能

运行方式:
    # 测试默认服务
    python test_sse.py

    # 测试指定的 SSE 服务

IUHHUI

参考文档

目录结构

.
├── Dockerfile
└── opt
    ├── startup.sh
    └── switchDmLog.sh

IUHHUI

Dockerfile

# 指定的官方 Flink 镜像作为基础
FROM flink:1.20.3-scala_2.12-java17

# Flink CDC 3.5.0 的 JAR 包下载地址 (以 MySQL、PostgreSQL、SQL Server 为例)
# 查找地址: https://github.com/apache/flink-cdc/releases/tag/release-3.5.0

# 下载 MySQL CDC 3.5.0 连接器

IUHHUI

要在 Vim 中搜索 'recv: ' 但排除 'recv: <'，使用：

/recv: <\@!

IUHHUI

import json
from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.hunyuan.v20230901 import hunyuan_client, models


def main(
    TENCENTCLOUD_SECRET_ID: str,
    TENCENTCLOUD_SECRET_KEY: str,

IUHHUI

import os


def main(
    DASHSCOPE_API_KEY: str,
    quick: str,
    text: str,
    source_lang: str,
):
    if "HOME" not in os.environ:

IUHHUI

import pdfplumber
import PyPDF2
import logging


def check_by_pypdf2(file_path):
    try:
        with open(file_path, "rb") as file:
            pdf_reader = PyPDF2.PdfReader(file)
            # 获取 PDF 的页数，尝试读取以确认文件是否损坏

IUHHUI

import PyPDF2

def check_pdf(file_path):
    try:
        with open(file_path, 'rb') as file:
            pdf_reader = PyPDF2.PdfReader(file)
            # 获取 PDF 的页数，尝试读取以确认文件是否损坏
            num_pages = len(pdf_reader.pages)
            print(f"文件 '{file_path}' 是有效的 PDF，包含 {num_pages} 页。")
            return True

IUHHUI

import os
from openai import OpenAI

STR_CHINESE = "Chinese"
STR_ENGLISH = "English"
MODEL_QWEN_MT_TURBO = "qwen-mt-turbo"
MODEL_QWEN_MT_PLUS = "qwen-mt-plus"


def detect_language(text) -> str:

IUHHUI

NVIDIA 例子

1. 创建collection
2. insert data
3. search data