mcp_knowledge_ops 0.1.0

Knowledge operations including pipelines, workflows, scheduling, and automated processing for curation, summarization, and pattern mining in the MCP ecosystem.

MCP Knowledge Ops

Support This Project

If you find this package useful, consider supporting ongoing development on PayPal.

Support makemind via PayPal

---

MCP Knowledge Package Family

• mcp_bundle: Bundle schema, loader, validator, and expression language for MCP ecosystem.
• mcp_fact_graph: Temporal knowledge graph with evidence-based fact management and summarization.
• mcp_skill: Skill definitions and runtime execution for AI capabilities.
• mcp_profile: Profile definitions for AI personas with template rendering and appraisal.
• mcp_knowledge_ops: Knowledge operations including pipelines, workflows, and scheduling.
• mcp_knowledge: Unified integration package for the complete knowledge system.

---

A powerful Dart package for knowledge operations including pipelines, workflows, scheduling, and automated knowledge processing. Part of the MakeMind MCP ecosystem.

Features

Core Features

• Pipelines: Define and execute multi-stage processing pipelines
• Workflows: Complex workflow orchestration with branching and conditions
• Scheduling: Automated job scheduling with cron-like expressions
• Operations: Built-in operations for curation, summarization, and pattern mining

Pipeline Model

• Stages: Sequential processing stages
• Checkpoints: Save and restore pipeline state
• Metrics: Execution metrics and monitoring
• Error Handling: Retry logic and error recovery

Workflow Model

• Actions: Discrete workflow actions
• Conditions: Conditional branching
• Variables: Workflow-scoped variables
• Events: Event-driven execution

Built-in Operations

• Curation: Candidate review and confirmation
• Summarization: Entity summary generation
• Pattern Mining: Pattern discovery in knowledge
• Index Refresh: Knowledge index maintenance

Quick Start

Installation

Add to your pubspec.yaml:

dependencies:
  mcp_knowledge_ops: ^0.1.0

Basic Usage

import 'package:mcp_knowledge_ops/mcp_knowledge_ops.dart';

void main() async {
  // Create pipeline executor
  final executor = PipelineExecutor();

  // Define a curation pipeline
  final pipeline = Pipeline(
    id: 'daily_curation',
    name: 'Daily Curation',
    stages: [
      PipelineStage(
        id: 'fetch_candidates',
        type: StageType.fetch,
        config: {'status': 'pending', 'limit': 100},
      ),
      PipelineStage(
        id: 'auto_confirm',
        type: StageType.process,
        config: {'threshold': 0.95},
      ),
      PipelineStage(
        id: 'notify',
        type: StageType.output,
        config: {'channel': 'slack'},
      ),
    ],
  );

  // Execute pipeline
  final result = await executor.execute(
    pipeline,
    CurationInput(entityIds: ['user_123']),
    PipelineContext(runId: 'run_001'),
  );

  print('Success: ${result.success}');
  print('Metrics: ${result.metrics}');
}

Core Concepts

Pipelines

Pipelines define sequential processing stages:

final pipeline = Pipeline(
  id: 'summarization_pipeline',
  name: 'Entity Summarization',
  stages: [
    PipelineStage(
      id: 'fetch_facts',
      type: StageType.fetch,
      config: {'domain': 'preferences'},
    ),
    PipelineStage(
      id: 'generate_summary',
      type: StageType.process,
      config: {'model': 'claude-3-5-sonnet'},
    ),
    PipelineStage(
      id: 'store_summary',
      type: StageType.output,
      config: {'collection': 'summaries'},
    ),
  ],
  checkpointEnabled: true,
  retryPolicy: RetryPolicy(maxRetries: 3),
);

Workflows

Workflows orchestrate complex multi-step processes:

final workflow = Workflow(
  id: 'onboarding_workflow',
  name: 'User Onboarding',
  actions: [
    WorkflowAction(
      id: 'extract_preferences',
      type: ActionType.skill,
      skillId: 'preference_extractor',
    ),
    WorkflowAction(
      id: 'check_premium',
      type: ActionType.condition,
      condition: '{{user.isPremium}}',
      onTrue: 'premium_setup',
      onFalse: 'basic_setup',
    ),
    WorkflowAction(
      id: 'premium_setup',
      type: ActionType.skill,
      skillId: 'premium_onboarding',
    ),
    WorkflowAction(
      id: 'basic_setup',
      type: ActionType.skill,
      skillId: 'basic_onboarding',
    ),
  ],
);

// Execute workflow
final workflowExecutor = WorkflowExecutor(handlers: handlers);
final execution = await workflowExecutor.execute(
  workflow,
  {'user': userContext},
);

Scheduling

Schedule automated jobs:

final scheduler = Scheduler(
  config: SchedulerConfig(
    timezone: 'UTC',
    maxConcurrentJobs: 5,
  ),
);

// Schedule daily curation
await scheduler.schedule(
  ScheduledJob(
    id: 'daily_curation',
    name: 'Daily Candidate Curation',
    schedule: CronSchedule('0 6 * * *'), // 6 AM daily
    pipeline: curationPipeline,
    input: CurationInput(),
  ),
);

// Schedule hourly summarization
await scheduler.schedule(
  ScheduledJob(
    id: 'hourly_summary',
    name: 'Hourly Summary Refresh',
    schedule: CronSchedule('0 * * * *'), // Every hour
    pipeline: summarizationPipeline,
    input: SummarizationInput(staleOnly: true),
  ),
);

// Start scheduler
await scheduler.start();

Built-in Operations

Curation

final curationPipeline = CurationPipeline(
  factGraph: factGraphPort,
);

final result = await executor.execute(
  curationPipeline,
  CurationInput(
    entityIds: ['user_123', 'user_456'],
    autoConfirmThreshold: 0.95,
  ),
  context,
);

print('Confirmed: ${result.output.confirmedCount}');
print('Rejected: ${result.output.rejectedCount}');

Summarization

final summarizationPipeline = SummarizationPipeline(
  factGraph: factGraphPort,
  llm: llmPort,
);

final result = await executor.execute(
  summarizationPipeline,
  SummarizationInput(
    entityIds: ['user_123'],
    summaryTypes: ['preferences', 'history'],
  ),
  context,
);

print('Updated: ${result.output.updatedSummaries}');

Pattern Mining

final patternPipeline = PatternMiningPipeline(
  factGraph: factGraphPort,
  llm: llmPort,
);

final result = await executor.execute(
  patternPipeline,
  PatternMiningInput(
    domain: 'preferences',
    minSupport: 0.1,
  ),
  context,
);

print('Patterns: ${result.output.patterns}');

Port-Based Architecture

The package uses a port-based dependency injection pattern:

// Define custom operation port
class MyOpsPort implements OpsPort {
  @override
  Future<void> executeOperation(Operation op) async {
    // Your implementation
  }
}

// Use custom ports
final executor = PipelineExecutor(
  opsPort: MyOpsPort(),
);

API Reference

PipelineExecutor

Method	Description
execute(pipeline, input, context)	Execute a pipeline
pause(runId)	Pause pipeline execution
resume(runId)	Resume paused pipeline
cancel(runId)	Cancel pipeline

WorkflowExecutor

Method	Description
execute(workflow, context)	Execute a workflow
getExecution(id)	Get execution status
listExecutions(filter)	List executions

Scheduler

Method	Description
schedule(job)	Schedule a job
unschedule(jobId)	Remove a job
start()	Start scheduler
stop()	Stop scheduler

Examples

Complete Examples Available

• example/basic_pipeline.dart - Basic pipeline execution
• example/workflow.dart - Workflow orchestration
• example/scheduling.dart - Job scheduling
• example/curation.dart - Curation pipeline
• example/summarization.dart - Summarization pipeline

Contributing

We welcome contributions! Please see our Contributing Guide for details.

Support

• Documentation
• Issue Tracker
• Discussions
• Support on Patreon

License

This project is licensed under the MIT License - see the LICENSE file for details.