@skynetxbt/core

SkynetXBT Agent Framework

A powerful graph-based autonomous agent framework for creating complex workflows with web3 integrations.

Overview

The SkynetXBT Agent Framework is a no-code flow builder that enables you to create sophisticated autonomous agents through visual graph-based workflows. The framework supports parallel and sequential execution, conditional branching, and seamless integration with 100s of web3 tools.

Core Concepts

Graph-Based Execution

Agents execute workflows as directed graphs where:

• Nodes represent individual flows (tasks/operations)
• Edges define execution order and dependencies
• Control Nodes manage flow control (entry, end, terminate)
• Parallel Execution allows multiple flows to run simultaneously
• Conditional Branching enables dynamic path selection

Flow Architecture

interface FlowConfig {
  name: string;
  description: string;
  plugins: string[];
  triggers: string[];
  isAutonomous?: boolean;
  abilities?: string[];
  examples?: string[];
}

interface IFlow {
  flowConfig: FlowConfig;
  execute(context: FlowContext): Promise<void | string | object>;
  executeAutonomous?(config: AutonomousFlowConfig): Promise<void>;
}

Quick Start

1. Create a Basic Flow

import { IFlow, FlowConfig, FlowContext } from '@skynetxbt/core';

export class CustomFlow implements IFlow {
  flowConfig: FlowConfig = {
    name: "CustomFlow",
    description: "A custom workflow for specific tasks",
    plugins: ["web3-plugin"],
    triggers: ["custom", "workflow", "task"]
  };

  async execute(context: FlowContext): Promise<void | string | object> {
    // Your flow logic here
    const { message, sessionId, userPublicKey } = context;
    
    // Process the input
    const result = await this.processData(message);
    
    return {
      success: true,
      data: result,
      timestamp: Date.now()
    };
  }

  private async processData(input: any) {
    // Implementation details
    return input;
  }
}

2. Create a Graph Configuration

import { GraphConfig } from '@skynetxbt/core';

const graphConfig: GraphConfig = {
  name: "sample-workflow",
  description: "A sample workflow demonstrating graph execution",
  flows: {
    "entry": {
      name: "entry",
      description: "Entry point",
      outNode: {
        flows: ["step1", "step2"],
        executionMode: "parallel",
        mergeStrategy: "array"
      }
    },
    "step1": {
      name: "FetchDataFlow",
      description: "Fetch external data",
      inNode: {
        flows: ["entry"]
      },
      outNode: {
        flows: ["step3"],
        executionMode: "sequential"
      }
    },
    "step2": {
      name: "ProcessDataFlow", 
      description: "Process user input",
      inNode: {
        flows: ["entry"]
      },
      outNode: {
        flows: ["step3"],
        executionMode: "sequential"
      }
    },
    "step3": {
      name: "MergeResultsFlow",
      description: "Combine results from parallel steps",
      inNode: {
        flows: ["step1", "step2"],
        mergeStrategy: "object"
      },
      outNode: {
        flows: ["end"]
      }
    },
    "end": {
      name: "end",
      description: "Workflow completion"
    }
  }
};

3. Initialize and Execute Agent

import { FlowAgent, LLMManager, FlowRegistry } from '@skynetxbt/core';

// Register your flows
const flowRegistry = FlowRegistry.getInstance();
flowRegistry.registerFlow(new CustomFlow(), "step1");
flowRegistry.registerFlow(new ProcessDataFlow(), "step2");
flowRegistry.registerFlow(new MergeResultsFlow(), "step3");

// Create LLM manager
const llmManager = new LLMManager({
  provider: 'openai',
  config: { model: 'gpt-4' }
});

// Create and execute flow agent
const agent = new FlowAgent(llmManager, graphConfig);

await agent.execute(
  "session-123",
  "user-public-key",
  "entry", // starting step
  { message: "Hello, process this data" }
);

Advanced Features

Conditional Branching

Flows can return a next property to control execution path:

async execute(context: FlowContext): Promise<any> {
  const condition = await this.evaluateCondition(context.message);
  
  return {
    data: "processed",
    next: condition ? 0 : 1 // Choose which output flow to execute
  };
}

Parallel Execution with Merge Strategies

// Graph configuration for parallel execution
outNode: {
  flows: ["flowA", "flowB", "flowC"],
  executionMode: "parallel",
  mergeStrategy: "object" // "concat", "object", or "array"
}

Autonomous Flows

Create self-executing flows that run on schedules:

export class AutonomousMonitorFlow implements IFlow {
  flowConfig: FlowConfig = {
    name: "AutonomousMonitorFlow",
    description: "Monitors blockchain events autonomously",
    plugins: ["blockchain-plugin"],
    triggers: ["monitor"],
    isAutonomous: true,
    autonomousConfig: {
      schedule: {
        type: "interval",
        value: 60000 // Run every minute
      },
      maxRuns: 100
    }
  };

  async execute(context: FlowContext) {
    // Monitor blockchain events
    const events = await this.checkBlockchainEvents();
    
    if (events.length > 0) {
      // Process events and take action
      return this.processEvents(events);
    }
    
    return { status: "no_events", timestamp: Date.now() };
  }

  async executeAutonomous(config: AutonomousFlowConfig) {
    // Autonomous execution logic
    console.log('Running autonomous monitoring...');
  }
}

Built-in Flows

The framework includes many pre-built flows:

• FetchUriFlow: HTTP requests with authentication
• ConditionalFlow: Dynamic path selection
• TimerFlow: Delayed execution
• TelegramFlow: Telegram bot integration
• WebhookFlow: HTTP webhook handling
• GoogleFlow: Google services integration
• BeefyFlow: DeFi yield farming
• DefilllamaFlow: DeFi protocol analysis

Event System

Monitor flow execution with built-in events:

import { FlowEventManager } from '@skynetxbt/core';

// Listen to flow events
FlowEventManager.on('flowExecutionStart', (data) => {
  console.log('Flow started:', data);
});

FlowEventManager.on('flowStepEnd', (data) => {
  console.log('Step completed:', data);
});

FlowEventManager.on('flowExecutionEnd', (data) => {
  console.log('Flow finished:', data);
});

Error Handling

The framework provides robust error handling:

• Step-level errors: Individual steps can fail without stopping the entire workflow
• Dependency resolution: Smart dependency checking prevents deadlocks
• Loop detection: Automatic detection and prevention of infinite loops
• Graceful degradation: Workflows continue execution when possible

API Reference

FlowAgent Methods

class FlowAgent {
  constructor(llmManager: LLMManager, graphConfig: GraphConfig)
  
  async execute(
    sessionId: string,
    userPublicKey: string, 
    startStepName?: string,
    initialPayload?: any
  ): Promise<void>
  
  setGraphConfiguration(graph: GraphConfig): void
  getGraphConfiguration(): GraphConfig | null
  getNextSteps(currentStepName: string): string[]
}

FlowRegistry Methods

class FlowRegistry {
  static getInstance(): FlowRegistry
  
  registerFlow(flow: IFlow, instanceId?: string): void
  registerFlows(flows: IFlow[]): void
  getFlow(name: string, instanceId?: string): IFlow | undefined
  getAllFlows(): Map<string, IFlow>
  
  async analyzeFlow(message: string): Promise<string | undefined>
}

Running the Demo

To see the framework in action:

cd example/demo-agent-flow
bun install
bun run start

Debugging

Shutdown Agent for Development

curl -X POST http://localhost:3010/kill \
  -H "Content-Type: application/json" \
  -d '{"shutdownKey": "hello"}'

Architecture Benefits

• Scalability: Graph-based execution scales to complex workflows
• Maintainability: Modular flows are easy to test and update
• Flexibility: Dynamic branching adapts to runtime conditions
• Performance: Parallel execution maximizes throughput
• Observability: Comprehensive event system for monitoring
• Error Resilience: Robust error handling keeps workflows running

Contributing

• Fork the repository
• Create your feature branch
• Implement your flow following the IFlow interface
• Add tests and documentation
• Submit a pull request