MCP 서버
중국어판
프로젝트 개요
FastAPI와 MCP(Model Context Protocol)를 기반으로 구축된 이 프로젝트는 AI 모델과 개발 환경 간의 표준화된 컨텍스트 상호작용을 지원합니다. 모델 배포를 간소화하고, 효율적인 API 엔드포인트를 제공하며, 모델 입력 및 출력의 일관성을 보장함으로써 AI 애플리케이션의 확장성과 유지 관리성을 향상시켜 개발자가 AI 작업을 보다 쉽게 통합하고 관리할 수 있도록 지원합니다.
MCP(Model Context Protocol)는 AI 모델과 개발 환경 간의 컨텍스트 상호작용을 위한 통합 프로토콜입니다. 이 프로젝트는 초기화, 샘플링, 세션 관리를 포함한 기본적인 MCP 프로토콜 기능을 지원하는 Python 기반 MCP 서버 구현을 제공합니다.
특징
- JSON-RPC 2.0 : 표준 JSON-RPC 2.0 프로토콜을 기반으로 한 요청-응답 통신
- SSE 연결 : 실시간 알림을 위한 서버 전송 이벤트 연결 지원
- 모듈형 디자인 : 쉬운 확장 및 사용자 정의를 위한 모듈형 아키텍처
- 비동기 처리 : FastAPI와 비동기 IO를 활용한 고성능 서비스
- 전체 클라이언트 : 전체 테스트 클라이언트 구현 포함
프로젝트 구조
지엑스피1
설치
- 저장소를 복제합니다.
git clone https://github.com/freedanfan/mcp_server.git
cd mcp_server
- 종속성 설치:
pip install -r requirements.txt
용법
서버 시작
기본적으로 서버는 127.0.0.1:12000 에서 시작됩니다. 환경 변수를 사용하여 호스트와 포트를 사용자 지정할 수 있습니다.
export MCP_SERVER_HOST=0.0.0.0
export MCP_SERVER_PORT=8000
python mcp_server.py
클라이언트 실행
다른 터미널에서 클라이언트를 실행합니다.
서버가 기본 주소에서 실행되지 않는 경우 환경 변수를 설정할 수 있습니다.
export MCP_SERVER_URL="http://your-server-address:port"
python mcp_client.py
API 엔드포인트
서버는 다음과 같은 API 엔드포인트를 제공합니다.
- 루트 경로 (
/ ): 서버 정보를 제공합니다. - API 엔드포인트 (
/api ): JSON-RPC 요청을 처리합니다. - SSE 엔드포인트 (
/sse ): SSE 연결을 처리합니다.
MCP 프로토콜 구현
초기화 흐름
- 클라이언트는 SSE를 통해 서버에 연결합니다.
- 서버는 API 엔드포인트 URI를 반환합니다.
- 클라이언트는 프로토콜 버전 및 기능을 포함하는 초기화 요청을 보냅니다.
- 서버는 초기화 요청에 응답하고 서버 기능을 반환합니다.
샘플링 요청
클라이언트는 다음과 같은 프롬프트를 사용하여 샘플링 요청을 보낼 수 있습니다.
{
"jsonrpc": "2.0",
"id": "request-id",
"method": "sample",
"params": {
"prompt": "Hello, please introduce yourself."
}
}
서버는 샘플링 결과를 반환합니다.
{
"jsonrpc": "2.0",
"id": "request-id",
"result": {
"content": "This is a response to the prompt...",
"usage": {
"prompt_tokens": 10,
"completion_tokens": 50,
"total_tokens": 60
}
}
}
세션 종료
클라이언트는 종료 요청을 보낼 수 있습니다.
{
"jsonrpc": "2.0",
"id": "request-id",
"method": "shutdown",
"params": {}
}
서버가 정상적으로 종료됩니다.
{
"jsonrpc": "2.0",
"id": "request-id",
"result": {
"status": "shutting_down"
}
}
개발 확장
새로운 방법 추가
새로운 MCP 메서드를 추가하려면 MCPServer 클래스에 핸들러 함수를 추가하고 _register_methods 메서드에 등록합니다.
def handle_new_method(self, params: dict) -> dict:
"""Handle new method"""
logger.info(f"Received new method request: {params}")
# Processing logic
return {"result": "success"}
def _register_methods(self):
# Register existing methods
self.router.register_method("initialize", self.handle_initialize)
self.router.register_method("sample", self.handle_sample)
self.router.register_method("shutdown", self.handle_shutdown)
# Register new method
self.router.register_method("new_method", self.handle_new_method)
AI 모델 통합
실제 AI 모델을 통합하려면 handle_sample 메서드를 수정하세요.
async def handle_sample(self, params: dict) -> dict:
"""Handle sampling request"""
logger.info(f"Received sampling request: {params}")
# Get prompt
prompt = params.get("prompt", "")
# Call AI model API
# For example: using OpenAI API
response = await openai.ChatCompletion.acreate(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
content = response.choices[0].message.content
usage = response.usage
return {
"content": content,
"usage": {
"prompt_tokens": usage.prompt_tokens,
"completion_tokens": usage.completion_tokens,
"total_tokens": usage.total_tokens
}
}
문제 해결
일반적인 문제
- 연결 오류 : 서버가 실행 중이고 클라이언트가 올바른 서버 URL을 사용하고 있는지 확인하세요.
- 405 메서드가 허용되지 않음 : 클라이언트가 올바른 API 엔드포인트에 요청을 보내고 있는지 확인하세요.
- SSE 연결 실패 : 네트워크 연결 및 방화벽 설정을 확인하세요
벌채 반출
서버와 클라이언트 모두 자세한 로깅을 제공합니다. 자세한 내용은 로그를 참조하세요.
# Increase log level
export PYTHONPATH=.
python -m logging -v DEBUG -m mcp_server
참고문헌
특허
이 프로젝트는 MIT 라이선스에 따라 라이선스가 부여됩니다. 자세한 내용은 라이선스 파일을 참조하세요.