-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathserver.py
More file actions
110 lines (99 loc) · 3.72 KB
/
server.py
File metadata and controls
110 lines (99 loc) · 3.72 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
import os
import sys
import mcp
import json
import asyncio
from contextlib import AsyncExitStack
from typing import List, Dict, Any, Optional, Annotated
from dotenv import load_dotenv
# from mcp.types import Tool, TextContent
import logging
load_dotenv('./.env')
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
logger = logging.getLogger(__name__)
class GitlabMCP:
GITLAB_ACCESS_TOKEN: str
GITLAB_PROJECT_ID: str
tools: List[Any]
is_conn: bool
session: Optional[mcp.ClientSession]
_session_ctx: Optional[Any]
exit_stack: AsyncExitStack
stdio: List[Any]
_stdio_ctx: Any
def __init__(self, GITLAB_ACCESS_TOKEN='', GITLAB_PROJECT_ID=''):
self.GITLAB_ACCESS_TOKEN = GITLAB_ACCESS_TOKEN
self.GITLAB_PROJECT_ID = GITLAB_PROJECT_ID
self.tools = []
self.is_conn = False
self.session = None
self._session_ctx = None
self.exit_stack = AsyncExitStack()
self.stdio = [None, None]
self._stdio_ctx = None
async def _connect(self):
try:
if not self.is_conn:
server_params = mcp.StdioServerParameters(
command='npx',
args=[
"-y",
"@zereight/mcp-gitlab"
],
env={
'GITLAB_PERSONAL_ACCESS_TOKEN': self.GITLAB_ACCESS_TOKEN,
'GITLAB_ALLOWED_PROJECT_IDS': self.GITLAB_PROJECT_ID
}
)
self._stdio_ctx = mcp.client.stdio.stdio_client(server_params)
self.stdio = await self._stdio_ctx.__aenter__()
self._session_ctx = mcp.ClientSession(*self.stdio)
self.session= await self._session_ctx.__aenter__()
await self.session.initialize()
self.is_conn = True
logger.info(f'Connected successfully !')
else:
logger.info(f'Connection already exists !')
except Exception as e:
logger.exception(f'Error connecting to unoff gitlab server: {e}')
async def disconnect(self):
try:
if self._session_ctx:
await self._session_ctx.__aexit__(None, None, None)
self._session_ctx = None
if self._stdio_ctx:
await self._stdio_ctx.__aexit__(None, None, None)
self.is_conn = False
logger.info(f'Connection closed !')
self.stdio = None
self.session = None
except Exception as e:
logger.exception(f'Error disconnecting from unoff gitlab server: {e}')
async def get_tools(self) -> List[Any]:
if self.is_conn:
if self.tools:
return self.tools
resp = await self.session.list_tools()
tools = resp.tools
self.tools = tools
return tools
logger.info(f'Not connected to the server !')
return []
async def call_tool(self, tool_name: str, args: Dict[str, Any]) -> Any:
if self.is_conn:
try:
res = await self.session.call_tool(tool_name, arguments=args)
return res
except Exception as e:
logger.exception(f'Error calling tool {tool_name}: {e}')
return 'Error! Try Again'
async def main():
gitlabMCP = GitlabMCP(os.environ['GITLAB_ACCESS_TOKEN'], os.environ['GITLAB_PROJECT_ID'])
await gitlabMCP._connect()
tools = await gitlabMCP.get_tools()
logger.info(f'Tools: {len(tools)}')
for tool in tools:
logger.info(f'Tool: {tool}')
await gitlabMCP.disconnect()
if __name__=='__main__':
asyncio.run(main())