Creating Agents - Ensemble Edge

Already read Your First Agent? This goes deeper.

Agent Structure

Agents follow a simple directory structure with separate YAML and TypeScript files:

agents/
└── my-agent/
    ├── my-agent.yaml      # Agent contract (WHAT it does)
    ├── my-agent.ts        # Handler implementation (HOW it does it)
    ├── my-agent.test.ts   # Tests (optional)
    └── README.md          # Documentation (optional)

The Handler Pattern

Conductor follows an interface vs implementation pattern:

YAML declares WHAT - The agent’s contract: inputs, outputs, and metadata
TypeScript defines HOW - The actual implementation logic

This keeps agents testable, type-safe, and debuggable while avoiding complex logic in YAML.

Simple Handler Example

Here’s a minimal agent showing the handler pattern:

YAML (Contract)
TypeScript (Implementation)

# agents/greeter/greeter.yaml
name: greeter
operation: code
handler: ./greeter.ts
description: Greets a user by name

schema:
  input:
    type: object
    properties:
      name:
        type: string
        description: Name to greet
  output:
    type: object
    properties:
      greeting:
        type: string
        description: Personalized greeting

// agents/greeter/greeter.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

interface GreeterInput {
  name: string
}

interface GreeterOutput {
  greeting: string
}

export default async function greeter(
  input: GreeterInput,
  ctx: AgentExecutionContext
): Promise<GreeterOutput> {
  // Access location for personalized greeting
  const timeOfDay = ctx.location?.getTimeOfDay() ?? 'day'
  const greetings = {
    morning: 'Good morning',
    afternoon: 'Good afternoon',
    evening: 'Good evening',
    day: 'Hello'
  }

  return {
    greeting: `${greetings[timeOfDay]}, ${input.name}!`
  }
}

Key Points:

The YAML handler: ./greeter.ts points to the TypeScript file
The TypeScript function signature: (input, ctx) => Promise<Output>
The function name can be anything (typically matches the agent name)
Full type safety with TypeScript interfaces
Access location and edge context for geo-aware behavior

Complete Agent Example

# agents/company-enricher/agent.yaml
agent: company-enricher
description: Enriches company data from multiple sources with fallbacks

inputs:
  company_name:
    type: string
    required: true
    description: Company name to enrich
  include_news:
    type: boolean
    default: false
    description: Whether to include recent news

cache:
  ttl: 86400  # Cache entire agent for 24 hours
  key: enrich-${input.company_name}

operations:
  # Step 1: Search for company
  - name: search
    operation: http
    config:
      url: https://api.duckduckgo.com/?q=${input.company_name}+official+website&format=json
      method: GET
    cache:
      ttl: 86400
      key: search-${input.company_name}
    retry:
      maxAttempts: 3
      backoff: exponential
      initialDelay: 1000

  # Step 2: Scrape website
  - name: scrape-primary
    operation: http
    config:
      url: ${search.output.AbstractURL}
      method: GET
      timeout: 10000
    retry:
      maxAttempts: 2

  # Step 3: Fallback scrape if primary fails
  - name: scrape-fallback
    condition: ${scrape-primary.failed}
    operation: http
    config:
      url: https://www.crunchbase.com/organization/${input.company_name}
      method: GET

  # Step 4: Extract with AI
  - name: extract
    operation: think
    config:
      provider: openai
      model: gpt-4o-mini
      temperature: 0.3
      maxTokens: 500
      prompt: |
        Extract company info from this HTML:
        ${scrape-primary.output?.body || scrape-fallback.output?.body}

        Return JSON with:
        {
          "name": "Company name",
          "description": "Brief description (1-2 sentences)",
          "industry": "Primary industry",
          "founded": "Year founded if available"
        }
    cache:
      ttl: 3600
      key: extract-${input.company_name}

  # Step 5: Optionally fetch news
  - name: fetch-news
    condition: ${input.include_news && extract.executed}
    operation: http
    config:
      url: https://newsapi.org/v2/everything?q=${input.company_name}&sortBy=publishedAt&pageSize=5
      headers:
        Authorization: Bearer ${env.NEWS_API_KEY}
    cache:
      ttl: 3600

  # Step 6: Store in cache
  - name: cache-result
    condition: ${extract.executed}
    operation: storage
    config:
      type: kv
      action: put
      key: company-${input.company_name}
      value:
        company_data: ${extract.output}
        news: ${fetch-news.output?.body?.articles}
        cached_at: ${Date.now()}
      expirationTtl: 86400

outputs:
  company_data: ${extract.output}
  news: ${fetch-news.output?.body?.articles}
  source_url: ${search.output.AbstractURL}
  from_cache: ${__cache_hit}

Caching Strategies

Agent-Level Caching

Cache the entire agent execution:

agent: expensive-agent

cache:
  ttl: 3600
  key: expensive-${input.id}

operations:
  # All operations run if cache miss
  # Skip all if cache hit

Operation-Level Caching

Cache specific operations:

operations:
  - name: expensive-scrape
    operation: http
    config:
      url: ${input.url}
    cache:
      ttl: 86400  # 24 hours
      key: scrape-${input.url}

  - name: ai-analysis
    operation: think
    config:
      prompt: Analyze: ${expensive-scrape.output}
    cache:
      ttl: 3600  # 1 hour
      key: analyze-${input.url}

Dynamic Cache Keys

operations:
  - name: cached-op
    operation: think
    config:
      prompt: ${input.text}
    cache:
      ttl: ${input.cache_duration || 3600}
      key: ${input.cache_key || `default-${input.text}`}

Retry Logic

Basic Retry

operations:
  - name: flaky-api
    operation: http
    config:
      url: https://api.example.com
    retry:
      maxAttempts: 3
      backoff: exponential
      initialDelay: 1000

Advanced Retry

operations:
  - name: smart-retry
    operation: http
    config:
      url: https://api.example.com
    retry:
      maxAttempts: 5
      backoff: exponential  # 1s, 2s, 4s, 8s, 16s
      initialDelay: 1000
      maxDelay: 30000       # Cap at 30s
      retryOn: [500, 502, 503, 504]  # Only retry these codes
      timeout: 10000        # Timeout per attempt

Conditional Retry

operations:
  - name: try-operation
    operation: http
    config:
      url: https://api.example.com
    retry:
      maxAttempts: 2

  - name: check-retry-count
    operation: code
    config:
      script: scripts/check-retry
    input:
      retryCount: ${try-operation.retry_count}

// scripts/check-retry.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default async function checkRetry(context: AgentExecutionContext) {
  const { retryCount } = context.input
  return {
    should_retry: retryCount < 3,
    count: retryCount
  }
}

Error Handling

Fallback Pattern

operations:
  - name: primary
    operation: http
    config:
      url: https://primary-api.com
    retry:
      maxAttempts: 2

  - name: secondary
    condition: ${primary.failed}
    operation: http
    config:
      url: https://backup-api.com

  - name: cached
    condition: ${primary.failed && secondary.failed}
    operation: storage
    config:
      type: kv
      action: get
      key: fallback-data

outputs:
  data: ${primary.output || secondary.output || cached.output}
  source: ${primary.executed ? 'primary' : secondary.executed ? 'secondary' : 'cache'}

Error Logging

operations:
  - name: risky-operation
    operation: http
    config:
      url: https://api.example.com

  - name: log-error
    condition: ${risky-operation.failed}
    operation: data
    config:
      backend: d1
      binding: DB
      operation: execute
      sql: |
        INSERT INTO error_log (agent, operation, error, timestamp)
        VALUES (?, ?, ?, ?)
      params:
        - company-enricher
        - risky-operation
        - ${risky-operation.error}
        - ${Date.now()}

Graceful Degradation

operations:
  - name: enhanced-analysis
    operation: think
    config:
      provider: openai
      model: gpt-4o
      prompt: ${input.text}

  - name: basic-analysis
    condition: ${enhanced-analysis.failed}
    operation: think
    config:
      provider: cloudflare
      model: '@cf/meta/llama-3.1-8b-instruct'
      prompt: ${input.text}

outputs:
  analysis: ${enhanced-analysis.output || basic-analysis.output}
  quality: ${enhanced-analysis.executed ? 'enhanced' : 'basic'}

Testing Agents

Unit Tests

// agents/company-enricher/agent.test.ts
import { describe, it, expect, beforeEach } from 'vitest';
import { TestConductor } from '@ensemble-edge/conductor/testing';

describe('company-enricher agent', () => {
  let conductor: TestConductor;

  beforeEach(async () => {
    conductor = await TestConductor.create();
    await conductor.loadProject('./');
  });

  it('should enrich company data', async () => {
    const result = await conductor.executeAgent('company-enricher', {
      company_name: 'Anthropic',
      include_news: false
    });

    expect(result).toBeSuccessful();
    expect(result.output.company_data).toHaveProperty('name');
    expect(result.output.company_data).toHaveProperty('description');
    expect(result.output.company_data).toHaveProperty('industry');
  });

  it('should include news when requested', async () => {
    const result = await conductor.executeAgent('company-enricher', {
      company_name: 'Anthropic',
      include_news: true
    });

    expect(result.output.news).toBeDefined();
    expect(Array.isArray(result.output.news)).toBe(true);
  });

  it('should use cache on second call', async () => {
    // First call
    const result1 = await conductor.executeAgent('company-enricher', {
      company_name: 'Anthropic'
    });

    // Second call
    const result2 = await conductor.executeAgent('company-enricher', {
      company_name: 'Anthropic'
    });

    expect(result1.output.from_cache).toBe(false);
    expect(result2.output.from_cache).toBe(true);
  });

  it('should handle scraping failures with fallback', async () => {
    // Mock primary scrape failure
    conductor.mockOperation('scrape-primary', {
      failed: true,
      error: 'Timeout'
    });

    const result = await conductor.executeAgent('company-enricher', {
      company_name: 'Anthropic'
    });

    expect(result).toBeSuccessful();
    expect(result.operations['scrape-fallback'].executed).toBe(true);
  });

  it('should retry failed operations', async () => {
    conductor.mockOperation('search', {
      failed: true,
      retry_count: 2
    });

    const result = await conductor.executeAgent('company-enricher', {
      company_name: 'Anthropic'
    });

    expect(result.operations.search.retry_count).toBeGreaterThan(0);
  });
});

Integration Tests

describe('company-enricher integration', () => {
  it('should work end-to-end with real APIs', async () => {
    const conductor = await TestConductor.create({
      env: {
        NEWS_API_KEY: process.env.NEWS_API_KEY
      }
    });
    await conductor.loadProject('./');

    const result = await conductor.executeAgent('company-enricher', {
      company_name: 'Anthropic',
      include_news: true
    });

    expect(result).toBeSuccessful();
    expect(result.output.company_data.name).toContain('Anthropic');
    expect(result.output.news.length).toBeGreaterThan(0);
    expect(result.output.source_url).toMatch(/https?:\/\/.+/);
  });
});

Mock Data

describe('company-enricher with mocks', () => {
  it('should handle mocked responses', async () => {
    const conductor = await TestConductor.create();
    await conductor.loadProject('./');

    // Mock search result
    conductor.mockOperation('search', {
      output: {
        AbstractURL: 'https://anthropic.com'
      }
    });

    // Mock scrape result
    conductor.mockOperation('scrape-primary', {
      output: {
        body: '<html><body>Anthropic is an AI safety company...</body></html>'
      }
    });

    const result = await conductor.executeAgent('company-enricher', {
      company_name: 'Anthropic'
    });

    expect(result).toBeSuccessful();
  });
});

Performance Optimization

Parallel Operations

operations:
  # These run in parallel (no dependencies)
  - name: fetch-a
    operation: http
    config:
      url: https://api-a.com

  - name: fetch-b
    operation: http
    config:
      url: https://api-b.com

  - name: fetch-c
    operation: http
    config:
      url: https://api-c.com

  # This waits for all 3
  - name: merge
    operation: code
    config:
      script: scripts/merge-api-results
    input:
      a: ${fetch-a.output}
      b: ${fetch-b.output}
      c: ${fetch-c.output}

// scripts/merge-api-results.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default async function mergeApiResults(context: AgentExecutionContext) {
  const { a, b, c } = context.input
  return { a, b, c }
}

Minimize Dependencies

# Bad: Sequential
operations:
  - name: step1
    operation: http
    config:
      url: https://api.com/step1

  - name: step2
    operation: http
    config:
      url: https://api.com/step2
      data: ${step1.output}  # Creates dependency

# Good: Parallel
operations:
  - name: step1
    operation: http
    config:
      url: https://api.com/step1

  - name: step2
    operation: http
    config:
      url: https://api.com/step2

  - name: combine
    operation: code
    config:
      script: scripts/combine-outputs
    input:
      step1: ${step1.output}
      step2: ${step2.output}

// scripts/combine-outputs.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default async function combineOutputs(context: AgentExecutionContext) {
  const { step1, step2 } = context.input
  return { ...step1, ...step2 }
}

Early Termination

operations:
  # Quick check first
  - name: quick-check
    operation: code
    config:
      script: scripts/quick-validate
    input:
      data: ${input.data}

  # Only run expensive ops if valid
  - name: expensive-analysis
    condition: ${quick-check.output.valid}
    operation: think
    config:
      provider: openai
      model: gpt-4o
      prompt: ${input.data}

// scripts/quick-validate.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default async function quickValidate(context: AgentExecutionContext) {
  const { data } = context.input
  return { valid: data && data.length > 0 }
}

Using Agents in TypeScript Ensembles

Once you’ve created YAML agents with TypeScript handlers, use them in TypeScript ensembles:

// ensembles/enrichment-pipeline.ts
import { createEnsemble, step, parallel, tryStep } from '@anthropic/conductor'

const enrichmentPipeline = createEnsemble('enrichment-pipeline')
  .setDescription('Enrich company data with fallbacks')

  // Use your custom agent
  .addStep(
    tryStep('enrich-safe')
      .try(
        step('enrich')
          .agent('company-enricher')  // Your YAML agent
          .input({
            company_name: '${input.company}',
            include_news: true
          })
          .cache({ ttl: 86400, key: 'enrich-${input.company}' })
          .retry({ maxAttempts: 3, backoff: 'exponential' })
      )
      .catch(
        step('fallback-enrich')
          .agent('basic-enricher')
          .input({ company_name: '${input.company}' })
      )
  )

  // Parallel data fetching
  .addStep(
    parallel('fetch-social')
      .steps(
        step('linkedin').agent('scraper').input({ url: '${input.linkedinUrl}' }),
        step('twitter').agent('scraper').input({ url: '${input.twitterUrl}' })
      )
  )

  // AI analysis
  .addStep(
    step('analyze')
      .operation('think')
      .config({
        provider: 'anthropic',
        model: 'claude-3-5-sonnet-20241022',
        prompt: `
          Analyze company data:
          Enriched: \${enrich.output}
          LinkedIn: \${linkedin.output}
          Twitter: \${twitter.output}
        `
      })
  )

  .build()

export default enrichmentPipeline

Validating Your Setup

Validate both agents and ensembles:

# Validate all agents
ensemble conductor validate agents/ -r

# Validate all ensembles (YAML and TypeScript)
ensemble conductor validate ensembles/ -r

# Validate everything
ensemble conductor validate . -r

For complete TypeScript API documentation, see the TypeScript API Reference.

Best Practices

Single Responsibility - One agent, one job
Cache Aggressively - Cache expensive operations
Handle Failures - Always have fallbacks
Use Retry Logic - For transient failures
Test Thoroughly - Unit and integration tests
Document Inputs/Outputs - Clear schemas
Monitor Performance - Track execution times
Version with Edgit - Track changes over time
Use Location Context - Leverage location for localization and compliance
Consider Edge Context - Use edge for security decisions

Next Steps

TypeScript API

Complete TypeScript reference

Writing Ensembles

Orchestrate agents

Location Context

Geographic data and compliance

Edge Context

Network and protocol info

​Agent Structure

​The Handler Pattern

​Simple Handler Example

​Complete Agent Example

​Caching Strategies

​Agent-Level Caching

​Operation-Level Caching

​Dynamic Cache Keys

​Retry Logic

​Basic Retry

​Advanced Retry

​Conditional Retry

​Error Handling

​Fallback Pattern

​Error Logging

​Graceful Degradation

​Testing Agents

​Unit Tests

​Integration Tests

​Mock Data

​Performance Optimization

​Parallel Operations

​Minimize Dependencies

​Early Termination

​Using Agents in TypeScript Ensembles

​Validating Your Setup

​Best Practices

​Next Steps

TypeScript API

Writing Ensembles

Location Context

Edge Context

Agent Structure

The Handler Pattern

Simple Handler Example

Complete Agent Example

Caching Strategies

Agent-Level Caching

Operation-Level Caching

Dynamic Cache Keys

Retry Logic

Basic Retry

Advanced Retry

Conditional Retry

Error Handling

Fallback Pattern

Error Logging

Graceful Degradation

Testing Agents

Unit Tests

Integration Tests

Mock Data

Performance Optimization

Parallel Operations

Minimize Dependencies

Early Termination

Using Agents in TypeScript Ensembles

Validating Your Setup

Best Practices

Next Steps