mirror of
https://github.com/Tencent/WeKnora.git
synced 2026-09-24 16:29:01 +08:00
feat: Enhance MCP service configuration with stdio transport support
- Added support for stdio transport type in MCP service configuration, allowing command execution via standard input/output. - Updated MCPService interface to include optional stdio_config and env_vars fields for managing command arguments and environment variables. - Enhanced frontend components to handle stdio transport, including dynamic form fields for command and arguments. - Implemented validation logic for stdio configuration in the service update process, ensuring proper handling of required fields. - Updated backend logic to manage stdio transport connections and ensure proper client lifecycle management.
This commit is contained in:
@@ -49,7 +49,10 @@ WORKDIR /app
|
||||
# Install runtime dependencies
|
||||
RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.tuna.tsinghua.edu.cn/g' /etc/apk/repositories && \
|
||||
apk update && apk upgrade && \
|
||||
apk add --no-cache build-base postgresql-client mysql-client ca-certificates tzdata sed curl bash vim wget
|
||||
apk add --no-cache build-base postgresql-client mysql-client ca-certificates tzdata sed curl bash vim wget \
|
||||
nodejs npm python3 py3-pip && \
|
||||
pip3 install --no-cache-dir pipx && \
|
||||
pipx install uvx
|
||||
|
||||
# Create a non-root user and switch to it
|
||||
RUN mkdir -p /data/files && \
|
||||
|
||||
@@ -6,8 +6,8 @@ export interface MCPService {
|
||||
name: string
|
||||
description: string
|
||||
enabled: boolean
|
||||
transport_type: 'sse' | 'http-streamable'
|
||||
url: string
|
||||
transport_type: 'sse' | 'http-streamable' | 'stdio'
|
||||
url?: string // Optional: required for SSE/HTTP Streamable
|
||||
headers?: Record<string, string>
|
||||
auth_config?: {
|
||||
api_key?: string
|
||||
@@ -19,6 +19,11 @@ export interface MCPService {
|
||||
retry_count?: number
|
||||
retry_delay?: number
|
||||
}
|
||||
stdio_config?: {
|
||||
command: 'uvx' | 'npx' // Command: uvx or npx
|
||||
args: string[] // Command arguments array
|
||||
}
|
||||
env_vars?: Record<string, string> // Environment variables for stdio transport
|
||||
created_at?: string
|
||||
updated_at?: string
|
||||
}
|
||||
|
||||
@@ -36,11 +36,11 @@
|
||||
<div class="service-name">
|
||||
{{ service.name }}
|
||||
<t-tag
|
||||
:theme="service.transport_type === 'sse' ? 'success' : 'primary'"
|
||||
:theme="getTransportTypeTheme(service.transport_type)"
|
||||
size="small"
|
||||
variant="light"
|
||||
>
|
||||
{{ service.transport_type === 'sse' ? 'SSE' : 'HTTP Streamable' }}
|
||||
{{ getTransportTypeLabel(service.transport_type) }}
|
||||
</t-tag>
|
||||
</div>
|
||||
<div class="service-controls">
|
||||
@@ -274,6 +274,34 @@ const handleMenuAction = (data: { value: string }, service: MCPService) => {
|
||||
}
|
||||
}
|
||||
|
||||
// Get transport type theme for tag
|
||||
const getTransportTypeTheme = (transportType: string) => {
|
||||
switch (transportType) {
|
||||
case 'sse':
|
||||
return 'success'
|
||||
case 'http-streamable':
|
||||
return 'primary'
|
||||
case 'stdio':
|
||||
return 'warning'
|
||||
default:
|
||||
return 'default'
|
||||
}
|
||||
}
|
||||
|
||||
// Get transport type label
|
||||
const getTransportTypeLabel = (transportType: string) => {
|
||||
switch (transportType) {
|
||||
case 'sse':
|
||||
return 'SSE'
|
||||
case 'http-streamable':
|
||||
return 'HTTP Streamable'
|
||||
case 'stdio':
|
||||
return 'Stdio'
|
||||
default:
|
||||
return transportType
|
||||
}
|
||||
}
|
||||
|
||||
onMounted(() => {
|
||||
loadServices()
|
||||
})
|
||||
|
||||
@@ -29,13 +29,101 @@
|
||||
<t-radio-group v-model="formData.transport_type">
|
||||
<t-radio value="sse">SSE (Server-Sent Events)</t-radio>
|
||||
<t-radio value="http-streamable">HTTP Streamable</t-radio>
|
||||
<t-radio value="stdio">Stdio</t-radio>
|
||||
</t-radio-group>
|
||||
</t-form-item>
|
||||
|
||||
<t-form-item label="服务 URL" name="url">
|
||||
<!-- URL for SSE/HTTP Streamable -->
|
||||
<t-form-item
|
||||
v-if="formData.transport_type !== 'stdio'"
|
||||
label="服务 URL"
|
||||
name="url"
|
||||
>
|
||||
<t-input v-model="formData.url" placeholder="https://example.com/mcp" />
|
||||
</t-form-item>
|
||||
|
||||
<!-- Stdio Config -->
|
||||
<template v-if="formData.transport_type === 'stdio'">
|
||||
<t-form-item label="命令" name="stdio_config.command">
|
||||
<t-radio-group v-model="formData.stdio_config.command">
|
||||
<t-radio value="uvx">uvx</t-radio>
|
||||
<t-radio value="npx">npx</t-radio>
|
||||
</t-radio-group>
|
||||
</t-form-item>
|
||||
|
||||
<t-form-item label="参数" name="stdio_config.args">
|
||||
<div class="args-input-container">
|
||||
<div
|
||||
v-for="(arg, index) in formData.stdio_config.args"
|
||||
:key="index"
|
||||
class="arg-item"
|
||||
>
|
||||
<t-input
|
||||
v-model="formData.stdio_config.args[index]"
|
||||
:placeholder="`参数 ${index + 1}`"
|
||||
class="arg-input"
|
||||
/>
|
||||
<t-button
|
||||
variant="text"
|
||||
theme="danger"
|
||||
@click="removeArg(index)"
|
||||
:disabled="formData.stdio_config.args.length === 1"
|
||||
>
|
||||
<template #icon><t-icon name="delete" /></template>
|
||||
</t-button>
|
||||
</div>
|
||||
<t-button
|
||||
variant="outline"
|
||||
size="small"
|
||||
@click="addArg"
|
||||
class="add-arg-btn"
|
||||
>
|
||||
<template #icon><t-icon name="add" /></template>
|
||||
添加参数
|
||||
</t-button>
|
||||
</div>
|
||||
</t-form-item>
|
||||
|
||||
<t-form-item label="环境变量">
|
||||
<div class="env-vars-container">
|
||||
<div
|
||||
v-for="(value, key, index) in formData.env_vars"
|
||||
:key="index"
|
||||
class="env-var-item"
|
||||
>
|
||||
<t-input
|
||||
v-model="envVarKeys[index]"
|
||||
placeholder="变量名"
|
||||
class="env-key-input"
|
||||
@blur="updateEnvVarKey(index, envVarKeys[index])"
|
||||
/>
|
||||
<t-input
|
||||
v-model="formData.env_vars[key]"
|
||||
placeholder="变量值"
|
||||
type="password"
|
||||
class="env-value-input"
|
||||
/>
|
||||
<t-button
|
||||
variant="text"
|
||||
theme="danger"
|
||||
@click="removeEnvVar(key)"
|
||||
>
|
||||
<template #icon><t-icon name="delete" /></template>
|
||||
</t-button>
|
||||
</div>
|
||||
<t-button
|
||||
variant="outline"
|
||||
size="small"
|
||||
@click="addEnvVar"
|
||||
class="add-env-var-btn"
|
||||
>
|
||||
<template #icon><t-icon name="add" /></template>
|
||||
添加环境变量
|
||||
</t-button>
|
||||
</div>
|
||||
</t-form-item>
|
||||
</template>
|
||||
|
||||
<t-form-item label="启用服务" name="enabled">
|
||||
<t-switch v-model="formData.enabled" />
|
||||
</t-form-item>
|
||||
@@ -122,8 +210,13 @@ const formData = ref({
|
||||
name: '',
|
||||
description: '',
|
||||
enabled: true,
|
||||
transport_type: 'sse' as 'sse' | 'http-streamable',
|
||||
transport_type: 'sse' as 'sse' | 'http-streamable' | 'stdio',
|
||||
url: '',
|
||||
stdio_config: {
|
||||
command: 'uvx' as 'uvx' | 'npx',
|
||||
args: ['']
|
||||
},
|
||||
env_vars: {} as Record<string, string>,
|
||||
auth_config: {
|
||||
api_key: '',
|
||||
token: ''
|
||||
@@ -135,12 +228,54 @@ const formData = ref({
|
||||
}
|
||||
})
|
||||
|
||||
// Track env var keys separately for easier editing
|
||||
const envVarKeys = ref<string[]>([])
|
||||
|
||||
const rules: Record<string, FormRule[]> = {
|
||||
name: [{ required: true, message: '请输入服务名称', type: 'error' }],
|
||||
transport_type: [{ required: true, message: '请选择传输类型', type: 'error' }],
|
||||
url: [
|
||||
{ required: true, message: '请输入服务 URL', type: 'error' },
|
||||
{ url: true, message: '请输入有效的 URL', type: 'error' }
|
||||
{
|
||||
validator: (val: string) => {
|
||||
if (formData.value.transport_type !== 'stdio') {
|
||||
if (!val || val.trim() === '') {
|
||||
return { result: false, message: '请输入服务 URL', type: 'error' }
|
||||
}
|
||||
// Basic URL validation
|
||||
try {
|
||||
new URL(val)
|
||||
return { result: true }
|
||||
} catch {
|
||||
return { result: false, message: '请输入有效的 URL', type: 'error' }
|
||||
}
|
||||
}
|
||||
return { result: true }
|
||||
}
|
||||
}
|
||||
],
|
||||
'stdio_config.command': [
|
||||
{
|
||||
validator: (val: string) => {
|
||||
if (formData.value.transport_type === 'stdio') {
|
||||
if (!val || (val !== 'uvx' && val !== 'npx')) {
|
||||
return { result: false, message: '请选择命令 (uvx 或 npx)', type: 'error' }
|
||||
}
|
||||
}
|
||||
return { result: true }
|
||||
}
|
||||
}
|
||||
],
|
||||
'stdio_config.args': [
|
||||
{
|
||||
validator: (val: string[]) => {
|
||||
if (formData.value.transport_type === 'stdio') {
|
||||
if (!val || val.length === 0 || val.every(arg => !arg || arg.trim() === '')) {
|
||||
return { result: false, message: '请至少输入一个参数', type: 'error' }
|
||||
}
|
||||
}
|
||||
return { result: true }
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -157,6 +292,11 @@ const resetForm = () => {
|
||||
enabled: true,
|
||||
transport_type: 'sse',
|
||||
url: '',
|
||||
stdio_config: {
|
||||
command: 'uvx',
|
||||
args: ['']
|
||||
},
|
||||
env_vars: {},
|
||||
auth_config: {
|
||||
api_key: '',
|
||||
token: ''
|
||||
@@ -167,9 +307,74 @@ const resetForm = () => {
|
||||
retry_delay: 1
|
||||
}
|
||||
}
|
||||
envVarKeys.value = []
|
||||
formRef.value?.clearValidate()
|
||||
}
|
||||
|
||||
// Watch transport_type to reset related fields
|
||||
watch(
|
||||
() => formData.value.transport_type,
|
||||
(newType) => {
|
||||
if (newType === 'stdio') {
|
||||
formData.value.url = ''
|
||||
if (!formData.value.stdio_config || formData.value.stdio_config.args.length === 0) {
|
||||
formData.value.stdio_config = {
|
||||
command: 'uvx',
|
||||
args: ['']
|
||||
}
|
||||
}
|
||||
} else {
|
||||
formData.value.stdio_config = {
|
||||
command: 'uvx',
|
||||
args: ['']
|
||||
}
|
||||
formData.value.env_vars = {}
|
||||
envVarKeys.value = []
|
||||
}
|
||||
formRef.value?.clearValidate()
|
||||
}
|
||||
)
|
||||
|
||||
// Args management
|
||||
const addArg = () => {
|
||||
formData.value.stdio_config.args.push('')
|
||||
}
|
||||
|
||||
const removeArg = (index: number) => {
|
||||
if (formData.value.stdio_config.args.length > 1) {
|
||||
formData.value.stdio_config.args.splice(index, 1)
|
||||
}
|
||||
}
|
||||
|
||||
// Env vars management
|
||||
const addEnvVar = () => {
|
||||
const key = `VAR_${Date.now()}`
|
||||
formData.value.env_vars[key] = ''
|
||||
envVarKeys.value.push(key)
|
||||
}
|
||||
|
||||
const removeEnvVar = (key: string) => {
|
||||
delete formData.value.env_vars[key]
|
||||
const index = envVarKeys.value.indexOf(key)
|
||||
if (index > -1) {
|
||||
envVarKeys.value.splice(index, 1)
|
||||
}
|
||||
}
|
||||
|
||||
const updateEnvVarKey = (index: number, newKey: string) => {
|
||||
const oldKey = envVarKeys.value[index]
|
||||
if (oldKey && oldKey !== newKey && formData.value.env_vars[oldKey] !== undefined) {
|
||||
const value = formData.value.env_vars[oldKey]
|
||||
delete formData.value.env_vars[oldKey]
|
||||
if (newKey && newKey.trim() !== '') {
|
||||
formData.value.env_vars[newKey] = value
|
||||
envVarKeys.value[index] = newKey
|
||||
} else {
|
||||
envVarKeys.value.splice(index, 1)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Watch service prop to initialize form
|
||||
watch(
|
||||
() => props.service,
|
||||
@@ -181,6 +386,11 @@ watch(
|
||||
enabled: service.enabled ?? true,
|
||||
transport_type: service.transport_type || 'sse',
|
||||
url: service.url || '',
|
||||
stdio_config: service.stdio_config || {
|
||||
command: 'uvx',
|
||||
args: ['']
|
||||
},
|
||||
env_vars: service.env_vars || {},
|
||||
auth_config: {
|
||||
api_key: service.auth_config?.api_key || '',
|
||||
token: service.auth_config?.token || ''
|
||||
@@ -191,6 +401,8 @@ watch(
|
||||
retry_delay: service.advanced_config?.retry_delay || 1
|
||||
}
|
||||
}
|
||||
// Initialize env var keys
|
||||
envVarKeys.value = Object.keys(formData.value.env_vars)
|
||||
} else {
|
||||
resetForm()
|
||||
}
|
||||
@@ -210,7 +422,6 @@ const handleSubmit = async () => {
|
||||
description: formData.value.description,
|
||||
enabled: formData.value.enabled,
|
||||
transport_type: formData.value.transport_type,
|
||||
url: formData.value.url,
|
||||
auth_config: {
|
||||
api_key: formData.value.auth_config.api_key || undefined,
|
||||
token: formData.value.auth_config.token || undefined
|
||||
@@ -218,6 +429,26 @@ const handleSubmit = async () => {
|
||||
advanced_config: formData.value.advanced_config
|
||||
}
|
||||
|
||||
// Add URL or stdio_config based on transport type
|
||||
if (formData.value.transport_type === 'stdio') {
|
||||
// Filter out empty args
|
||||
const args = formData.value.stdio_config.args.filter(arg => arg && arg.trim() !== '')
|
||||
data.stdio_config = {
|
||||
command: formData.value.stdio_config.command,
|
||||
args: args.length > 0 ? args : undefined
|
||||
}
|
||||
// Filter out empty env vars
|
||||
const envVars: Record<string, string> = {}
|
||||
for (const [key, value] of Object.entries(formData.value.env_vars)) {
|
||||
if (key && key.trim() !== '' && value && value.trim() !== '') {
|
||||
envVars[key] = value
|
||||
}
|
||||
}
|
||||
data.env_vars = Object.keys(envVars).length > 0 ? envVars : undefined
|
||||
} else {
|
||||
data.url = formData.value.url || undefined
|
||||
}
|
||||
|
||||
if (props.mode === 'add') {
|
||||
await createMCPService(data)
|
||||
MessagePlugin.success('MCP 服务已创建')
|
||||
@@ -243,3 +474,49 @@ const handleClose = () => {
|
||||
}
|
||||
</script>
|
||||
|
||||
<style scoped lang="less">
|
||||
.args-input-container {
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 8px;
|
||||
|
||||
.arg-item {
|
||||
display: flex;
|
||||
gap: 8px;
|
||||
align-items: center;
|
||||
|
||||
.arg-input {
|
||||
flex: 1;
|
||||
}
|
||||
}
|
||||
|
||||
.add-arg-btn {
|
||||
align-self: flex-start;
|
||||
}
|
||||
}
|
||||
|
||||
.env-vars-container {
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 8px;
|
||||
|
||||
.env-var-item {
|
||||
display: flex;
|
||||
gap: 8px;
|
||||
align-items: center;
|
||||
|
||||
.env-key-input {
|
||||
width: 150px;
|
||||
}
|
||||
|
||||
.env-value-input {
|
||||
flex: 1;
|
||||
}
|
||||
}
|
||||
|
||||
.add-env-var-btn {
|
||||
align-self: flex-start;
|
||||
}
|
||||
}
|
||||
</style>
|
||||
|
||||
|
||||
@@ -73,6 +73,18 @@ func (t *MCPTool) Execute(ctx context.Context, args map[string]interface{}) (*ty
|
||||
}, nil
|
||||
}
|
||||
|
||||
// For stdio transport, ensure connection is released after use
|
||||
isStdio := t.service.TransportType == types.MCPTransportStdio
|
||||
if isStdio {
|
||||
defer func() {
|
||||
if err := client.Disconnect(); err != nil {
|
||||
logger.GetLogger(ctx).Warnf("Failed to disconnect stdio MCP client: %v", err)
|
||||
} else {
|
||||
logger.GetLogger(ctx).Infof("Stdio MCP client disconnected after tool execution")
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Call the tool via MCP
|
||||
result, err := client.CallTool(ctx, t.mcpTool.Name, args)
|
||||
if err != nil {
|
||||
@@ -194,6 +206,16 @@ func RegisterMCPTools(ctx context.Context, registry *ToolRegistry, services []*t
|
||||
continue
|
||||
}
|
||||
|
||||
// For stdio transport, ensure connection is released after listing tools
|
||||
isStdio := service.TransportType == types.MCPTransportStdio
|
||||
if isStdio {
|
||||
defer func() {
|
||||
if err := client.Disconnect(); err != nil {
|
||||
logger.GetLogger(ctx).Warnf("Failed to disconnect stdio MCP client after listing tools: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// List tools from the service with timeout
|
||||
// Create a new context with timeout for this specific operation
|
||||
listCtx, cancel := context.WithTimeout(ctx, listToolsTimeout)
|
||||
|
||||
@@ -108,8 +108,14 @@ func (r *mcpServiceRepository) Update(ctx context.Context, service *types.MCPSer
|
||||
if service.TransportType != "" {
|
||||
updateMap["transport_type"] = service.TransportType
|
||||
}
|
||||
if service.URL != "" {
|
||||
updateMap["url"] = service.URL
|
||||
if service.URL != nil {
|
||||
updateMap["url"] = *service.URL
|
||||
}
|
||||
if service.StdioConfig != nil {
|
||||
updateMap["stdio_config"] = service.StdioConfig
|
||||
}
|
||||
if service.EnvVars != nil {
|
||||
updateMap["env_vars"] = service.EnvVars
|
||||
}
|
||||
if service.Headers != nil {
|
||||
updateMap["headers"] = service.Headers
|
||||
|
||||
@@ -127,9 +127,15 @@ func (s *mcpServiceService) UpdateMCPService(ctx context.Context, service *types
|
||||
if service.TransportType != "" {
|
||||
existing.TransportType = service.TransportType
|
||||
}
|
||||
if service.URL != "" {
|
||||
if service.URL != nil {
|
||||
existing.URL = service.URL
|
||||
}
|
||||
if service.StdioConfig != nil {
|
||||
existing.StdioConfig = service.StdioConfig
|
||||
}
|
||||
if service.EnvVars != nil {
|
||||
existing.EnvVars = service.EnvVars
|
||||
}
|
||||
if service.Headers != nil {
|
||||
existing.Headers = service.Headers
|
||||
}
|
||||
@@ -149,9 +155,19 @@ func (s *mcpServiceService) UpdateMCPService(ctx context.Context, service *types
|
||||
return fmt.Errorf("failed to update MCP service: %w", err)
|
||||
}
|
||||
|
||||
// Check if critical configuration changed (URL, transport type, or auth config)
|
||||
// Check if critical configuration changed (URL/StdioConfig, transport type, or auth config)
|
||||
configChanged := false
|
||||
if service.URL != "" && service.URL != existing.URL {
|
||||
if service.URL != nil && existing.URL != nil && *service.URL != *existing.URL {
|
||||
configChanged = true
|
||||
} else if (service.URL != nil) != (existing.URL != nil) {
|
||||
configChanged = true
|
||||
}
|
||||
if service.StdioConfig != nil && existing.StdioConfig != nil {
|
||||
if service.StdioConfig.Command != existing.StdioConfig.Command ||
|
||||
!equalStringSlices(service.StdioConfig.Args, existing.StdioConfig.Args) {
|
||||
configChanged = true
|
||||
}
|
||||
} else if (service.StdioConfig != nil) != (existing.StdioConfig != nil) {
|
||||
configChanged = true
|
||||
}
|
||||
if service.TransportType != "" && service.TransportType != existing.TransportType {
|
||||
@@ -322,3 +338,16 @@ func (s *mcpServiceService) GetMCPServiceResources(ctx context.Context, tenantID
|
||||
|
||||
return resources, nil
|
||||
}
|
||||
|
||||
// equalStringSlices compares two string slices for equality
|
||||
func equalStringSlices(a, b []string) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
for i := range a {
|
||||
if a[i] != b[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -151,8 +151,34 @@ func (h *MCPServiceHandler) UpdateMCPService(c *gin.Context) {
|
||||
if transportType, ok := updateData["transport_type"].(string); ok {
|
||||
service.TransportType = types.MCPTransportType(transportType)
|
||||
}
|
||||
if url, ok := updateData["url"].(string); ok {
|
||||
service.URL = url
|
||||
if url, ok := updateData["url"].(string); ok && url != "" {
|
||||
service.URL = &url
|
||||
} else if _, exists := updateData["url"]; exists {
|
||||
// Explicitly set to nil if provided as null/empty
|
||||
service.URL = nil
|
||||
}
|
||||
if stdioConfig, ok := updateData["stdio_config"].(map[string]interface{}); ok {
|
||||
config := &types.MCPStdioConfig{}
|
||||
if command, ok := stdioConfig["command"].(string); ok {
|
||||
config.Command = command
|
||||
}
|
||||
if args, ok := stdioConfig["args"].([]interface{}); ok {
|
||||
config.Args = make([]string, len(args))
|
||||
for i, arg := range args {
|
||||
if str, ok := arg.(string); ok {
|
||||
config.Args[i] = str
|
||||
}
|
||||
}
|
||||
}
|
||||
service.StdioConfig = config
|
||||
}
|
||||
if envVars, ok := updateData["env_vars"].(map[string]interface{}); ok {
|
||||
service.EnvVars = make(types.MCPEnvVars)
|
||||
for k, v := range envVars {
|
||||
if str, ok := v.(string); ok {
|
||||
service.EnvVars[k] = str
|
||||
}
|
||||
}
|
||||
}
|
||||
if headers, ok := updateData["headers"].(map[string]interface{}); ok {
|
||||
service.Headers = make(types.MCPHeaders)
|
||||
|
||||
+35
-3
@@ -95,7 +95,10 @@ func NewMCPClient(config *ClientConfig) (MCPClient, error) {
|
||||
var err error
|
||||
switch config.Service.TransportType {
|
||||
case types.MCPTransportSSE:
|
||||
mcpClient, err = client.NewSSEMCPClient(config.Service.URL,
|
||||
if config.Service.URL == nil || *config.Service.URL == "" {
|
||||
return nil, fmt.Errorf("URL is required for SSE transport")
|
||||
}
|
||||
mcpClient, err = client.NewSSEMCPClient(*config.Service.URL,
|
||||
client.WithHTTPClient(httpClient),
|
||||
client.WithHeaders(headers),
|
||||
)
|
||||
@@ -103,14 +106,38 @@ func NewMCPClient(config *ClientConfig) (MCPClient, error) {
|
||||
return nil, fmt.Errorf("failed to create SSE client: %w", err)
|
||||
}
|
||||
case types.MCPTransportHTTPStreamable:
|
||||
if config.Service.URL == nil || *config.Service.URL == "" {
|
||||
return nil, fmt.Errorf("URL is required for HTTP Streamable transport")
|
||||
}
|
||||
// For HTTP streamable, we need to use transport options
|
||||
mcpClient, err = client.NewStreamableHttpClient(config.Service.URL,
|
||||
mcpClient, err = client.NewStreamableHttpClient(*config.Service.URL,
|
||||
transport.WithHTTPBasicClient(httpClient),
|
||||
transport.WithHTTPHeaders(headers),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create HTTP streamable client: %w", err)
|
||||
}
|
||||
case types.MCPTransportStdio:
|
||||
if config.Service.StdioConfig == nil {
|
||||
return nil, fmt.Errorf("stdio_config is required for stdio transport")
|
||||
}
|
||||
|
||||
// Convert env vars map to []string format (KEY=value)
|
||||
envVars := make([]string, 0, len(config.Service.EnvVars))
|
||||
for key, value := range config.Service.EnvVars {
|
||||
envVars = append(envVars, fmt.Sprintf("%s=%s", key, value))
|
||||
}
|
||||
|
||||
// Create stdio client with options
|
||||
// NewStdioMCPClientWithOptions(command string, env []string, args []string, opts ...transport.StdioOption)
|
||||
mcpClient, err = client.NewStdioMCPClientWithOptions(
|
||||
config.Service.StdioConfig.Command,
|
||||
envVars,
|
||||
config.Service.StdioConfig.Args,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create stdio client: %w", err)
|
||||
}
|
||||
default:
|
||||
return nil, ErrUnsupportedTransport
|
||||
}
|
||||
@@ -133,7 +160,12 @@ func (c *mcpGoClient) Connect(ctx context.Context) error {
|
||||
}
|
||||
|
||||
c.connected = true
|
||||
logger.GetLogger(ctx).Infof("MCP client connected to %s", c.service.URL)
|
||||
if c.service.TransportType == types.MCPTransportStdio {
|
||||
logger.GetLogger(ctx).Infof("MCP stdio client connected: %s %v",
|
||||
c.service.StdioConfig.Command, c.service.StdioConfig.Args)
|
||||
} else {
|
||||
logger.GetLogger(ctx).Infof("MCP client connected to %s", *c.service.URL)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
+48
-2
@@ -35,13 +35,20 @@ func NewMCPManager() *MCPManager {
|
||||
}
|
||||
|
||||
// GetOrCreateClient gets an existing client or creates a new one
|
||||
// For stdio transport, always creates a new client (not cached)
|
||||
// For SSE/HTTP Streamable, caches and reuses existing connections
|
||||
func (m *MCPManager) GetOrCreateClient(service *types.MCPService) (MCPClient, error) {
|
||||
// Check if service is enabled
|
||||
if !service.Enabled {
|
||||
return nil, fmt.Errorf("MCP service %s is not enabled", service.Name)
|
||||
}
|
||||
|
||||
// Check if client already exists
|
||||
// For stdio transport, always create a new client (don't cache)
|
||||
if service.TransportType == types.MCPTransportStdio {
|
||||
return m.createStdioClient(service)
|
||||
}
|
||||
|
||||
// For SSE/HTTP Streamable, check if client already exists and reuse
|
||||
m.clientsMu.RLock()
|
||||
client, exists := m.clients[service.ID]
|
||||
m.clientsMu.RUnlock()
|
||||
@@ -95,13 +102,52 @@ func (m *MCPManager) GetOrCreateClient(service *types.MCPService) (MCPClient, er
|
||||
return nil, fmt.Errorf("failed to initialize MCP client: %w", err)
|
||||
}
|
||||
|
||||
// Store client
|
||||
// Store client (only for non-stdio transports)
|
||||
m.clients[service.ID] = client
|
||||
|
||||
logger.GetLogger(m.ctx).Infof("MCP client created and initialized for service: %s", service.Name)
|
||||
return client, nil
|
||||
}
|
||||
|
||||
// createStdioClient creates a new stdio client (not cached)
|
||||
func (m *MCPManager) createStdioClient(service *types.MCPService) (MCPClient, error) {
|
||||
// Create new client
|
||||
config := &ClientConfig{
|
||||
Service: service,
|
||||
}
|
||||
|
||||
client, err := NewMCPClient(config)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create stdio MCP client: %w", err)
|
||||
}
|
||||
|
||||
// For stdio, Connect() starts the subprocess
|
||||
// Use manager's context for the connection lifecycle
|
||||
if err := client.Connect(m.ctx); err != nil {
|
||||
return nil, fmt.Errorf("failed to connect to stdio MCP service: %w", err)
|
||||
}
|
||||
|
||||
// Initialize needs a timeout to prevent hanging
|
||||
initTimeout := 30 * time.Second
|
||||
if service.AdvancedConfig != nil && service.AdvancedConfig.Timeout > 0 {
|
||||
initTimeout = time.Duration(service.AdvancedConfig.Timeout) * time.Second
|
||||
if initTimeout > 60*time.Second {
|
||||
initTimeout = 60 * time.Second
|
||||
}
|
||||
}
|
||||
|
||||
initCtx, initCancel := context.WithTimeout(m.ctx, initTimeout)
|
||||
defer initCancel()
|
||||
|
||||
if _, err := client.Initialize(initCtx); err != nil {
|
||||
client.Disconnect()
|
||||
return nil, fmt.Errorf("failed to initialize stdio MCP client: %w", err)
|
||||
}
|
||||
|
||||
logger.GetLogger(m.ctx).Infof("MCP stdio client created and initialized for service: %s", service.Name)
|
||||
return client, nil
|
||||
}
|
||||
|
||||
// GetClient gets an existing client
|
||||
func (m *MCPManager) GetClient(serviceID string) (MCPClient, bool) {
|
||||
m.clientsMu.RLock()
|
||||
|
||||
+54
-1
@@ -15,6 +15,7 @@ type MCPTransportType string
|
||||
const (
|
||||
MCPTransportSSE MCPTransportType = "sse" // Server-Sent Events
|
||||
MCPTransportHTTPStreamable MCPTransportType = "http-streamable" // HTTP Streamable
|
||||
MCPTransportStdio MCPTransportType = "stdio" // Stdio (Standard Input/Output)
|
||||
)
|
||||
|
||||
// MCPService represents an MCP (Model Context Protocol) service configuration
|
||||
@@ -25,10 +26,12 @@ type MCPService struct {
|
||||
Description string `json:"description" gorm:"type:text"`
|
||||
Enabled bool `json:"enabled" gorm:"default:true;index"`
|
||||
TransportType MCPTransportType `json:"transport_type" gorm:"type:varchar(50);not null"`
|
||||
URL string `json:"url" gorm:"type:varchar(512);not null"`
|
||||
URL *string `json:"url,omitempty" gorm:"type:varchar(512)"` // Optional: required for SSE/HTTP Streamable
|
||||
Headers MCPHeaders `json:"headers" gorm:"type:json"`
|
||||
AuthConfig *MCPAuthConfig `json:"auth_config" gorm:"type:json"`
|
||||
AdvancedConfig *MCPAdvancedConfig `json:"advanced_config" gorm:"type:json"`
|
||||
StdioConfig *MCPStdioConfig `json:"stdio_config,omitempty" gorm:"type:json"` // Required for stdio transport
|
||||
EnvVars MCPEnvVars `json:"env_vars,omitempty" gorm:"type:json"` // Environment variables for stdio
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
DeletedAt gorm.DeletedAt `json:"deleted_at" gorm:"index"`
|
||||
@@ -51,6 +54,15 @@ type MCPAdvancedConfig struct {
|
||||
RetryDelay int `json:"retry_delay"` // Delay between retries in seconds, default: 1
|
||||
}
|
||||
|
||||
// MCPStdioConfig represents stdio transport configuration
|
||||
type MCPStdioConfig struct {
|
||||
Command string `json:"command"` // Command: "uvx" or "npx"
|
||||
Args []string `json:"args"` // Command arguments array
|
||||
}
|
||||
|
||||
// MCPEnvVars represents environment variables as a map
|
||||
type MCPEnvVars map[string]string
|
||||
|
||||
// MCPTool represents a tool exposed by an MCP service
|
||||
type MCPTool struct {
|
||||
Name string `json:"name"`
|
||||
@@ -143,6 +155,47 @@ func (c *MCPAdvancedConfig) Scan(value interface{}) error {
|
||||
return json.Unmarshal(b, c)
|
||||
}
|
||||
|
||||
// Value implements driver.Valuer interface for MCPStdioConfig
|
||||
func (c *MCPStdioConfig) Value() (driver.Value, error) {
|
||||
if c == nil {
|
||||
return nil, nil
|
||||
}
|
||||
return json.Marshal(c)
|
||||
}
|
||||
|
||||
// Scan implements sql.Scanner interface for MCPStdioConfig
|
||||
func (c *MCPStdioConfig) Scan(value interface{}) error {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
b, ok := value.([]byte)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return json.Unmarshal(b, c)
|
||||
}
|
||||
|
||||
// Value implements driver.Valuer interface for MCPEnvVars
|
||||
func (e MCPEnvVars) Value() (driver.Value, error) {
|
||||
if e == nil {
|
||||
return nil, nil
|
||||
}
|
||||
return json.Marshal(e)
|
||||
}
|
||||
|
||||
// Scan implements sql.Scanner interface for MCPEnvVars
|
||||
func (e *MCPEnvVars) Scan(value interface{}) error {
|
||||
if value == nil {
|
||||
*e = nil
|
||||
return nil
|
||||
}
|
||||
b, ok := value.([]byte)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return json.Unmarshal(b, e)
|
||||
}
|
||||
|
||||
// GetDefaultAdvancedConfig returns default advanced configuration
|
||||
func GetDefaultAdvancedConfig() *MCPAdvancedConfig {
|
||||
return &MCPAdvancedConfig{
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
-- 06-add-mcp-stdio-support.sql
|
||||
-- Add stdio transport support for MCP services
|
||||
|
||||
-- Add stdio_config and env_vars columns
|
||||
ALTER TABLE mcp_services
|
||||
ADD COLUMN stdio_config JSON COMMENT 'Stdio configuration: {command: "uvx"/"npx", args: [...]}',
|
||||
ADD COLUMN env_vars JSON COMMENT 'Environment variables for stdio transport';
|
||||
|
||||
-- Make url column optional (remove NOT NULL constraint)
|
||||
ALTER TABLE mcp_services
|
||||
MODIFY COLUMN url VARCHAR(512) NULL;
|
||||
|
||||
-- Add check constraint: stdio transport requires stdio_config, others require url
|
||||
ALTER TABLE mcp_services
|
||||
ADD CONSTRAINT chk_mcp_transport_config CHECK (
|
||||
(transport_type = 'stdio' AND stdio_config IS NOT NULL) OR
|
||||
(transport_type != 'stdio' AND url IS NOT NULL)
|
||||
);
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
-- 06-add-mcp-stdio-support.sql
|
||||
-- Add stdio transport support for MCP services
|
||||
|
||||
-- Add stdio_config and env_vars columns
|
||||
ALTER TABLE mcp_services
|
||||
ADD COLUMN IF NOT EXISTS stdio_config JSONB, -- Stdio configuration: {command: "uvx"/"npx", args: [...]}
|
||||
ADD COLUMN IF NOT EXISTS env_vars JSONB; -- Environment variables for stdio transport
|
||||
|
||||
-- Make url column optional (remove NOT NULL constraint)
|
||||
ALTER TABLE mcp_services
|
||||
ALTER COLUMN url DROP NOT NULL;
|
||||
|
||||
-- Add check constraint: stdio transport requires stdio_config, others require url
|
||||
ALTER TABLE mcp_services
|
||||
ADD CONSTRAINT chk_mcp_transport_config CHECK (
|
||||
(transport_type = 'stdio' AND stdio_config IS NOT NULL) OR
|
||||
(transport_type != 'stdio' AND url IS NOT NULL)
|
||||
);
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
-- 06-add-mcp-stdio-support.sql
|
||||
-- Rollback stdio transport support for MCP services
|
||||
|
||||
BEGIN;
|
||||
|
||||
-- Remove check constraint
|
||||
ALTER TABLE mcp_services
|
||||
DROP CONSTRAINT IF EXISTS chk_mcp_transport_config;
|
||||
|
||||
-- Make url column required again
|
||||
ALTER TABLE mcp_services
|
||||
ALTER COLUMN url SET NOT NULL;
|
||||
|
||||
-- Remove stdio_config and env_vars columns
|
||||
ALTER TABLE mcp_services
|
||||
DROP COLUMN IF EXISTS env_vars,
|
||||
DROP COLUMN IF EXISTS stdio_config;
|
||||
|
||||
COMMIT;
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
-- 06-add-mcp-stdio-support.sql
|
||||
-- Add stdio transport support for MCP services
|
||||
|
||||
BEGIN;
|
||||
|
||||
-- Add stdio_config and env_vars columns
|
||||
ALTER TABLE mcp_services
|
||||
ADD COLUMN IF NOT EXISTS stdio_config JSONB, -- Stdio configuration: {command: "uvx"/"npx", args: [...]}
|
||||
ADD COLUMN IF NOT EXISTS env_vars JSONB; -- Environment variables for stdio transport
|
||||
|
||||
-- Make url column optional (remove NOT NULL constraint)
|
||||
ALTER TABLE mcp_services
|
||||
ALTER COLUMN url DROP NOT NULL;
|
||||
|
||||
-- Add check constraint: stdio transport requires stdio_config, others require url
|
||||
ALTER TABLE mcp_services
|
||||
ADD CONSTRAINT chk_mcp_transport_config CHECK (
|
||||
(transport_type = 'stdio' AND stdio_config IS NOT NULL) OR
|
||||
(transport_type != 'stdio' AND url IS NOT NULL)
|
||||
);
|
||||
|
||||
COMMIT;
|
||||
|
||||
Reference in New Issue
Block a user