个人资料协议服务器

一个提供统一访问个人专业资料(包括经验、出版物、职业时间线和社交媒体存在)的模型上下文协议(MCP)服务器。

  • by harisgulzar1
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 harisgulzar1/self-mcp

使用说明

自我MCP

一种模型上下文协议(MCP)服务器,提供对个人专业资料的统一访问,包括经验、出版物、职业时间表和社交媒体存在。

特性

  • 简介概述:获取全面的背景信息
  • 经验数据:获取详细的工作经验和专业历史
  • 出版物:检索科学出版物和会议演示文稿
  • 职业发展时间表:获得职业发展和里程碑
  • 社交媒体整合:访问LinkedIn、Instagram、Facebook和YouTube个人资料
  • 内容搜索:搜索所有个人资料内容
  • LLM集成:适用于免费LLM端点(Ollama、Hugging Face)

安装

  1. 克隆或下载代码文件

  2. 安装依赖项:

pip install -r requirements.txt
  1. 可选:安装软件包:
pip install -e .

快速开始

运行MCP服务器

python haris_profile_server.py

服务器使用stdio传输运行,并通过JSON-RPC消息进行通信。

运行客户端

python mcp_client.py

这将启动一个交互式聊天界面,您可以在其中询问有关Haris Gulzar个人资料的问题。

LLM集成选项

客户端支持多个免费LLM选项:

1.Ollama(推荐-当地)

在本地安装Olama:

# Install Ollama from https://ollama.ai/
ollama pull llama2  # or any other model

2.抱脸推理API(自由层)

  • 基本使用无需设置
  • 可选:从获取免费API令牌https://huggingface.co/settings/tokens
  • 如果您想要更高的速率限制,请在客户端代码中设置令牌

3.其他免费LLM API

您可以轻松地扩展客户端以使用其他免费的LLM服务:

  • 一起AI
  • 希腊
  • OpenRouter(免费层)
  • 本地模型通过calla.cpp、oobabooga等。

可用工具

MCP服务器提供以下工具:

  1. get_file_overview:专业概况和背景
  2. 获取经验:工作经验和职业经历
  3. get_出版物:科学出版物和会议
  4. get_care_timeline:职业发展和里程碑
  5. 获取社交链接:社交媒体简介和链接
  6. 搜索_文件_内容:搜索所有个人资料内容

使用示例

交互式聊天示例

You: Tell me about Haris Gulzar's background
You: What is his work experience?
You: Show me his publications
You: What are his social media profiles?
You: Search for machine learning in his profile

程序化使用

from mcp_client import HarisProfileClient

client = HarisProfileClient()
await client.connect_to_server()

# Get overview
overview = await client.call_tool("get_profile_overview")
print(overview)

# Search for specific content
results = await client.call_tool("search_profile_content", {"query": "research"})
print(results)

部署选项

1.地方发展

如上所示,在本地运行服务器和客户端。

2.Docker部署

创建一个 Dockerfile:

FROM python:3.9-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["python", "haris_profile_server.py"]

构建并运行:

docker build -t haris-profile-mcp .
docker run -p 8000:8000 haris-profile-mcp

3.云部署

MCP服务器可以部署到:

  • Heroku:创建一个 Procfile 随着 worker: python haris_profile_server.py
  • 铁路:从Git存储库直接部署
  • 谷歌云运行:使用上面的Dockerfile
  • AWS Lambda:需要适应无服务器环境

配置

环境变量

创建一个 .env 文件:

# Optional LLM API keys
OPENAI_API_KEY=your_key_here
ANTHROPIC_API_KEY=your_key_here
HUGGINGFACE_API_KEY=your_key_here

# Server configuration
SERVER_PORT=8000
LOG_LEVEL=INFO

# Timeout settings
HTTP_TIMEOUT=30
MCP_TIMEOUT=60

自定义配置

您可以通过修改中的配置文件URL和社交媒体链接来自定义服务器 haris_profile_server.py:

# Profile data URLs
self.profile_urls = {
    "overview": "https://sites.google.com/view/haris-gulzar/home",
    "experience": "https://sites.google.com/view/haris-gulzar/experience", 
    "publications": "https://sites.google.com/view/haris-gulzar/publications",
    "career_timeline": "https://sites.google.com/view/haris-gulzar/career-timeline"
}

# Social media URLs
self.social_urls = {
    "linkedin": "https://www.linkedin.com/in/haris-gulzar/",
    "instagram": "https://www.instagram.com/japanviaharis/",
    "facebook": "https://www.facebook.com/mharisgulzar/",
    "youtube": "https://www.youtube.com/@japanviaharis"
}

测试

测试MCP服务器

# Test server connectivity
python -c "
import asyncio
from mcp_client import HarisProfileClient

async def test():
    client = HarisProfileClient()
    connected = await client.connect_to_server()
    print('Connected:', connected)
    if connected:
        tools = await client.get_available_tools()
        print('Available tools:', len(tools))
        for tool in tools:
            print(f'  - {tool[\"name\"]}')

asyncio.run(test())
"

测试单个工具

python -c "
import asyncio
from mcp_client import HarisProfileClient

async def test_tool():
    client = HarisProfileClient()
    await client.connect_to_server()
    
    # Test overview tool
    result = await client.call_tool('get_profile_overview')
    print('Overview result length:', len(result))
    print('First 200 chars:', result[:200])

asyncio.run(test_tool())
"

扩展服务器

添加新工具

要添加新工具,请修改 _setup_handlers 方法in HarisProfileServer:

Tool(
    name="get_certifications",
    description="Fetch professional certifications",
    inputSchema={
        "type": "object",
        "properties": {
            "category": {
                "type": "string",
                "description": "Certification category"
            }
        },
        "required": []
    }
)

并添加相应的处理程序:

elif name == "get_certifications":
    category = arguments.get("category", "all") if arguments else "all"
    return await self._get_certifications(category)

添加新数据源

  1. 将URL添加到 profile_urls 或创建新类别
  2. 实现一个获取方法
  3. 添加工具定义
  4. 更新提示模板

自定义内容解析

为了更好地从特定网站提取内容,请修改 _fetch_and_parse_content 方法:

async def _fetch_and_parse_content(self, url: str) -> str:
    """Enhanced content parsing for specific sites"""
    response = await self.client.get(url)
    soup = BeautifulSoup(response.text, 'html.parser')
    
    # Site-specific parsing logic
    if 'sites.google.com' in url:
        return self._parse_google_sites(soup)
    elif 'linkedin.com' in url:
        return self._parse_linkedin(soup)
    # Add more site-specific parsers
    
    return self._parse_generic(soup)

与其他MCP客户端集成

此服务器适用于任何兼容MCP的客户端:

Claude桌面集成

添加到您的 claude_desktop_config.json:

{
  "mcpServers": {
    "haris-profile": {
      "command": "python",
      "args": ["/path/to/haris_profile_server.py"]
    }
  }
}

自定义MCP客户端

from mcp import ClientSession
from mcp.client.stdio import stdio_client

async def custom_client():
    session = await stdio_client(StdioServerParameters(
        command="python",
        args=["haris_profile_server.py"]
    ))
    
    # List available tools
    tools = await session.list_tools()
    
    # Call a tool
    result = await session.call_tool("get_profile_overview")
    
    return result

监控和日志记录

启用详细日志记录

import logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('haris_profile_server.log'),
        logging.StreamHandler()
    ]
)

健康检查端点

对于生产部署,考虑添加健康检查:

from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/health")
async def health_check():
    return {"status": "healthy", "server": "haris-profile-mcp"}

# Run alongside MCP server
if __name__ == "__main__":
    # Start FastAPI in background
    import threading
    threading.Thread(
        target=lambda: uvicorn.run(app, host="0.0.0.0", port=8001)
    ).start()
    
    # Start MCP server
    asyncio.run(main())

故障排除

常见问题

  1. 服务器无法启动:

    • 检查Python版本(需要3.8+)
    • 验证是否已安装所有依赖项
    • 检查端口冲突
  2. 内容获取失败:

    • 验证URL是否可访问
    • 检查网络连接
    • 查看超时设置
  3. LLM集成不起作用:

    • 对于Ollama:确保服务正在运行(ollama serve)
    • 对于HuggingFace:检查API限制和令牌
    • 验证对API端点的网络访问
  4. 大型配置文件的内存问题:

    • 实现内容分块
    • 添加缓存机制
    • 限制解析器中的内容大小

调试模式

运行调试日志:

PYTHONPATH=. python -c "
import logging
logging.basicConfig(level=logging.DEBUG)
import asyncio
from haris_profile_server import main
asyncio.run(main())
"

性能优化

缓存

添加缓存以减少重复请求:

import asyncio
from functools import lru_cache
from datetime import datetime, timedelta

class CachedContent:
    def __init__(self, content: str, timestamp: datetime):
        self.content = content
        self.timestamp = timestamp
    
    def is_expired(self, ttl_minutes: int = 60) -> bool:
        return datetime.now() - self.timestamp > timedelta(minutes=ttl_minutes)

# Add to HarisProfileServer class
self.content_cache = {}

async def _fetch_and_parse_content_cached(self, url: str) -> str:
    if url in self.content_cache and not self.content_cache[url].is_expired():
        return self.content_cache[url].content
    
    content = await self._fetch_and_parse_content(url)
    self.content_cache[url] = CachedContent(content, datetime.now())
    return content

异步优化

同时获取多个源:

async def _get_all_profile_data(self) -> Dict[str, str]:
    """Fetch all profile data concurrently"""
    tasks = [
        self._fetch_and_parse_content(url) 
        for url in self.profile_urls.values()
    ]
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return {
        section: result if isinstance(result, str) else f"Error: {result}"
        for section, result in zip(self.profile_urls.keys(), results)
    }

贡献

  1. 分叉存储库
  2. 创建要素分支
  3. 添加您的改进
  4. 彻底测试
  5. 提交拉取请求

开发设置

# Clone repository
git clone <repository-url>
cd haris-profile-mcp-server

# Install development dependencies
pip install -r requirements.txt
pip install -e .

# Run tests
python -m pytest tests/

# Format code
black *.py

许可证

该项目根据MIT许可证获得许可。有关详细信息,请参阅LICENSE文件。

支持

如有疑问或问题:

  • 在存储库中创建问题
  • 联系人:harisgulzar@outlook.com
  • 文档:请参阅README和代码注释

此MCP服务器为将个人/专业档案数据与AI助手集成提供了基础。根据您的特定需求扩展和定制它。