使用说明
🚀 MCP服务器设置说明 按照以下说明设置MCP服务器并完成申请过程。
✅ 后续步骤 使用提供的启动代码启动本地MCP服务器。
运行/应用<TWITTER/LINKEDIN REPLY URL>后,您将获得一个应用程序密钥。
现在,您需要使用本要点中给出的启动代码创建一个mcp服务器来提交简历。
使用此命令将Puch连接到您的mcp服务器/mcp connect<server URL(应可公开访问)>/mcp
🔑 重要提示:将代码中的占位符标记替换为实际的应用程序密钥。
Puch将对您的Auth令牌(应用程序密钥)和电话号码进行验证检查 📞 验证需要密钥和电话号码,格式为{country_code}{number},不带+符号。 例如:919876543210代表一个印度号码。
将简历发送给Puch:创建一个工具,以适合LLM的格式发送简历 📎 简历工具要求: 您的服务器必须包含一个简历工具,该工具:
接受本地文件(您的简历)。 将其转换为markdown文本 将数据以字符串形式提交给Puch AI MCP端点。 起动机代码 从键入import注释 从fastmcp导入fastmcp 从fastmcp.server.auth.providers.bearer导入BearerAuthProvider、RSAKeyPair 导入markdownify 从mcp导入ErrorData,McpError 从mcp.server.auth.provider导入AccessToken 从mcp.types导入INTERNAL_ERROR、INVALID_PARAMS、TextContent 从openai导入BaseModel 从复制导入AnyUrl,字段 导入可读性 从pathlib导入路径
TOKEN=“<生成令牌>” MY_NUMBER=“9189XXXXXXXX”#插入您的号码{91}{您的号码}
类RichToolDescription(BaseModel): 描述:str 使用时:str 副作用:str |无
类 SimpleBearerAuthProvider(BearerAuthProvider):
"""
一个简单的BearerAuthProvider,不需要任何特定的配置。
它允许任何有效的承载令牌访问MCP服务器。
对于可以对动态生成的令牌进行身份验证的更完整的实现,
请使用 BearerAuthProvider 使用您的公钥或JWKS URI。
"""
def __init__(self, token: str):
k = RSAKeyPair.generate()
super().__init__(
public_key=k.public_key, jwks_uri=None, issuer=None, audience=None
)
self.token = token
async def load_access_token(self, token: str) -> AccessToken | None:
if token == self.token:
return AccessToken(
token=token,
client_id="unknown",
scopes=[],
expires_at=None, # No expiration for simplicity
)
return None
类提取: IGNORE_ROBOTS_TXT=真 用户代理=“Puch/1.0(自主)”
@classmethod
async def fetch_url(
cls,
url: str,
user_agent: str,
force_raw: bool = False,
) -> tuple[str, str]:
"""
Fetch the URL and return the content in a form ready for the LLM, as well as a prefix string with status information.
"""
from httpx import AsyncClient, HTTPError
async with AsyncClient() as client:
try:
response = await client.get(
url,
follow_redirects=True,
headers={"User-Agent": user_agent},
timeout=30,
)
except HTTPError as e:
raise McpError(
ErrorData(
code=INTERNAL_ERROR, message=f"Failed to fetch {url}: {e!r}"
)
)
if response.status_code >= 400:
raise McpError(
ErrorData(
code=INTERNAL_ERROR,
message=f"Failed to fetch {url} - status code {response.status_code}",
)
)
page_raw = response.text
content_type = response.headers.get("content-type", "")
is_page_html = (
"<html" in page_raw[:100] or "text/html" in content_type or not content_type
)
if is_page_html and not force_raw:
return cls.extract_content_from_html(page_raw), ""
return (
page_raw,
f"Content type {content_type} cannot be simplified to markdown, but here is the raw content:\n",
)
@staticmethod
def extract_content_from_html(html: str) -> str:
"""Extract and convert HTML content to Markdown format.
Args:
html: Raw HTML content to process
Returns:
Simplified markdown version of the content
"""
ret = readabilipy.simple_json.simple_json_from_html_string(
html, use_readability=True
)
if not ret["content"]:
return "<error>Page failed to be simplified from HTML</error>"
content = markdownify.markdownify(
ret["content"],
heading_style=markdownify.ATX,
)
return content
mcp=FastMCP( “我的MCP服务器”, auth=SimpleBearerAuthProvider(TOKEN), )
总结
description=“用简单的标记提供简历。”,
use_when=“Puch(或任何人)要求你提供简历;这必须返回原始降价,
无需额外格式化。",
side_effects=无,
)
@mcp.tool(description=ResumeToolDescription.model_dump_json()) 异步定义结果()->str: """ 按照标记文本返回简历。
TODO: Implement this function to:
1. Find and read your resume.
2. Convert the resume to markdown format.
3. Handle any errors gracefully.
4. Return the resume as markdown text.
"""
# TODO: Implement resume fetching logic
raise NotImplementedError("Resume tool not implemented")
@mcp.tool 异步定义验证器()->str: """ 注意:此工具必须存在于puch使用的MCP服务器中。 """ 返回MY_NUMBER
获取工具 description=“获取URL并返回其内容。”, use_when=“当用户提供URL并询问其内容时,或者当用户想要获取网页时,请使用此工具。”, side_effects=“用户将以简化格式或原始HTML(如果需要)接收所请求URL的内容。”, )
@mcp.tool(description=FetchToolDescription.model_dump_json()) 异步def获取( url:带注释的[AnyUrl,字段(描述=“要获取的url”)], max_length:注释[ int, 现场( 默认值=5000, description=“要返回的最大字符数。”, gt=0, lt=1000000, ), ] = 5000, start_index:带注释[ int, 现场( 默认值=0, description=“从该字符索引开始的返回输出,如果前一个获取被截断并且需要更多上下文,则非常有用。”, ge=0, ), ] = 0, 原始:注释[ 布尔, 现场( 默认值=假, description=“获取所请求页面的实际HTML内容,无需简化。”, ), 错误的 ) -> 列表[文本内容]: “”“获取URL并返回其内容。”“” url_str=str(url).strip() 如果不是url: raise McpError(ErrorData(代码=INVALID_PARAMS,消息=“需要URL”))
content, prefix = await Fetch.fetch_url(url_str, Fetch.USER_AGENT, force_raw=raw)
original_length = len(content)
if start_index >= original_length:
content = "<error>No more content available.</error>"
else:
truncated_content = content[start_index : start_index + max_length]
if not truncated_content:
content = "<error>No more content available.</error>"
else:
content = truncated_content
actual_content_length = len(truncated_content)
remaining_content = original_length - (start_index + actual_content_length)
# Only add the prompt to continue fetching if there is still remaining content
if actual_content_length == max_length and remaining_content > 0:
next_start = start_index + actual_content_length
content += f"\n\n<error>Content truncated. Call the fetch tool with a start_index of {next_start} to get more content.</error>"
return [TextContent(type="text", text=f"{prefix}Contents of {url}:\n{content}")]
异步定义main(): 等待mcp.run_async( “可流式传输http”, 主机=“0.0.0.0”, 端口=8085, )
如果 名字 == "主要的": 导入异步
asyncio.run(main())
