Infor MCP Server

Infor SyteLine的统一AI代理编排器

一个全面的工作流编排系统，通过MCP（模型上下文协议）与Infor SyteLine集成，并使用GPT-5 Mini提供统一的AI代理协调。

项目结构

InforMCP Server/
├── orchestrator/                    # AI Orchestrator Core
│   ├── ai_orchestrator.py          # Main orchestrator engine
│   ├── workflow_coordinator.py     # Advanced coordination patterns
│   ├── orchestrator_interface.py   # Web dashboard and API
│   └── workflow_examples.py        # Pre-built workflow templates
├── mcp/                            # MCP Server Components
│   ├── app.py                      # Flask MCP server
│   ├── tools/                      # MCP tool implementations
│   │   ├── get_data.py
│   │   ├── upload_data.py
│   │   └── update_data.py
│   └── config/                     # Server configuration
├── utils/                          # Shared utilities
│   ├── simple_ido_client.py        # SyteLine API client
│   └── ido_integration.py          # IDO integration helpers
├── tests/                          # Test suites
│   ├── multi_workflow_test.py      # Multi-workflow testing
│   └── test_mcp_server.py          # MCP server testing
├── templates/                      # Web interface templates
│   └── dashboard.html              # Orchestrator dashboard
└── docs/                           # Documentation
    └── workflow_guide.md           # Workflow creation guide

快速开始

1.启动MCP服务器

python app.py

服务器运行于：http://localhost:8000

2.启动编排器界面

python orchestrator_interface.py

仪表板位于：http://localhost:5000

创建自己的工作流

基本工作流结构

工作流被定义为一系列具有依赖关系的步骤：

from ai_orchestrator import AIOrchestrator, WorkflowType

orchestrator = AIOrchestrator()

workflow_id = orchestrator.create_workflow(
    name="My Custom Workflow",
    workflow_type=WorkflowType.DATA_SYNC,
    description="My custom business process",
    steps_config=[
        {
            "name": "step_1",
            "description": "First step description",
            "action_type": "mcp_get_data",  # or "ai_reasoning", "file_processing"
            "parameters": {
                "ido_name": "SLCustomers",
                "properties": ["CustNum", "Name"]
            },
            "dependencies": [],  # Steps this depends on
            "timeout": 60,      # Seconds
            "retry_count": 3    # Retry attempts
        },
        {
            "name": "step_2",
            "description": "Second step with AI reasoning",
            "action_type": "ai_reasoning",
            "parameters": {
                "prompt": "Analyze the customer data",
                "context": {"analysis_type": "customer_segmentation"}
            },
            "dependencies": ["step_1"],  # Runs after step_1 completes
            "timeout": 120,
            "retry_count": 2
        }
    ]
)

# Start the workflow
orchestrator.start_workflow(workflow_id)

可用操作类型

1. mcp_get_data -从SyteLine检索数据

   parameters = {
       "ido_name": "SLCustomers",
       "properties": ["CustNum", "Name", "City"],
       "filter_criteria": "City = 'New York'"
   }

1. mcp_upload_data -在SyteLine中创建新记录

   parameters = {
       "ido_name": "SLCustomers",
       "data": {
           "Name": "Acme Corp",
           "City": "New York",
           "Status": "Active"
       }
   }

1. mcp_update_data -更新现有记录

   parameters = {
       "ido_name": "SLCustomers",
       "data": {"Status": "Inactive"},
       "filter_criteria": "City = 'Chicago'"
   }

1. ai_reasoning -使用GPT-5 Mini进行分析

   parameters = {
       "prompt": "Analyze this customer data and identify trends",
       "context": {"data_source": "customers"}
   }

1. file_processing -处理文件操作

   parameters = {
       "file_path": "data/invoices.pdf",
       "operation": "exists"  # or "read"
   }

使用预构建示例

from workflow_examples import WorkflowExamples

examples = WorkflowExamples(orchestrator)

# Customer onboarding
customer_data = {
    "name": "Acme Corporation",
    "address": "123 Business St",
    "city": "New York",
    "credit_limit": 50000
}
workflow_id = examples.create_customer_onboarding_workflow(customer_data)

# Purchase order processing
po_data = {
    "vendor_name": "Supplier Inc",
    "order_date": "2024-01-15",
    "items": [{"item_code": "WIDGET001", "quantity": 100}]
}
workflow_id = examples.create_purchase_order_workflow(po_data)

# Inventory reconciliation
workflow_id = examples.create_inventory_reconciliation_workflow("MAIN")

高级协调模式

顺序链

from workflow_coordinator import WorkflowCoordinator

coordinator = WorkflowCoordinator(orchestrator)

# Create workflows
workflow1 = orchestrator.create_workflow(...)
workflow2 = orchestrator.create_workflow(...)
workflow3 = orchestrator.create_workflow(...)

# Create sequential chain
chain_id = coordinator.create_sequential_workflow_chain(
    [workflow1, workflow2, workflow3],
    "My Sequential Process"
)

# Start the chain (only first workflow starts automatically)
coordinator.start_workflow_group(chain_id)

扇出/扇入模式

master_workflow = orchestrator.create_workflow(...)
worker1 = orchestrator.create_workflow(...)
worker2 = orchestrator.create_workflow(...)
aggregator = orchestrator.create_workflow(...)

pattern_id = coordinator.create_fan_out_fan_in_pattern(
    master_workflow,
    [worker1, worker2],
    aggregator,
    "Parallel Processing Pattern"
)

错误恢复

from workflow_coordinator import CoordinationRule, CoordinationEvent

# Create rule to start recovery workflow on failure
recovery_rule = CoordinationRule(
    id="failure_recovery",
    name="Failure Recovery Rule",
    trigger_event=CoordinationEvent.WORKFLOW_FAILED,
    trigger_conditions={'workflow_id': main_workflow},
    actions=[{
        'type': 'start_workflow',
        'workflow_id': recovery_workflow
    }]
)

coordinator.create_coordination_rule(recovery_rule)

API终点

工作流管理

• GET /api/workflows -列出所有工作流
• POST /api/workflows -创建新工作流
• GET /api/workflows/{id} -获取工作流状态
• POST /api/workflows/{id}/start -启动工作流
• POST /api/workflows/{id}/stop -停止工作流
• DELETE /api/workflows/{id} -删除工作流

模板和示例

• GET /api/templates/{type} -获取工作流模板
• POST /api/invoice-processing -创建发票处理工作流

统计

• GET /api/stats -获取编排器统计信息

监控与控制

Web仪表板

访问仪表板http://localhost:5000致：

• 实时查看所有活动工作流
• 监控工作流进度和状态
• 启动/停止工作流
• 查看执行统计信息
• 访问工作流模板

程序化监控

# Get workflow status
status = orchestrator.get_workflow_status(workflow_id)
print(f"Status: {status['status']}")
print(f"Current Step: {status['current_step'] + 1}/{status['total_steps']}")
print(f"Results: {status['results']}")

# List all workflows
workflows = orchestrator.list_workflows()
for workflow in workflows:
    print(f"{workflow['name']}: {workflow['status']}")

# Get coordination status
coord_status = coordinator.get_coordination_status()
print(f"Active dependencies: {coord_status['active_dependencies']}")

测试您的工作流程

运行综合测试

python multi_workflow_test.py

测试单个场景

tester = MultiWorkflowTester()

# Test concurrent execution
tester.test_basic_concurrent_workflows()

# Test coordination patterns
tester.test_sequential_coordination()
tester.test_fan_out_fan_in_pattern()

# Test error handling
tester.test_error_handling_and_recovery()

最佳实践

1. 简单开始：从2-3步工作流程开始，逐渐增加复杂性
2. 使用依赖关系：定义明确的步骤依赖关系，以确保正确的执行顺序
3. 设置超时：始终为每个步骤设置合理的超时
4. 实施检索：对可能暂时失败的步骤使用retry_count
5. 监控进度：使用仪表板监视工作流执行
6. 测试协调：生产使用前测试协调模式
7. 处理错误：始终包括错误恢复机制

SyteLine集成的常见IDO

• 客户: SLCustomers, SLCustomerDefaults, SLCustomerHistory
• 供应商: SLVendors, SLVendorDefaults
• 库存: SLItemLocs, SLItemForecasts, SLItemMasters
• 订单: SLSoItems, SLPoItems
• 财务数据: SLArOpenItems, SLApOpenItems

故障排除

常见问题

1. 工作流未启动：检查MCP服务器是否在端口8000上运行
2. 步骤超时：增加超时值或优化步骤逻辑
3. 依赖性问题：确保依赖关系中的步骤名称完全匹配
4. MCP连接错误：验证SyteLine凭据和网络连接

调试模式

通过设置日志级别启用调试日志记录：

import logging
logging.basicConfig(level=logging.DEBUG)

后续步骤

1. 探索中的预构建示例 workflow_examples.py
2. 测试协调模式 multi_workflow_test.py
3. 使用web仪表板监视工作流
4. 根据需要扩展自定义IDO集成
5. 为复杂场景实施额外的协调规则