Decorator that transforms Python functions into deterministic skills with automatic REST API endpoints. Skills are designed for business logic, calculations, data processing, and external integrations that require consistent, predictable behavior.

Basic Example

from agentfield import Agent

app = Agent(node_id="data-processor")

@app.skill
def calculate_metrics(data: list[float]) -> dict:
    """Calculate statistical metrics from data."""
    return {
        "mean": sum(data) / len(data),
        "min": min(data),
        "max": max(data),
        "count": len(data)
    }

curl -X POST http://localhost:8080/api/v1/execute/data-processor.calculate_metrics \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "data": [10.5, 20.3, 15.7, 30.2, 25.1]
    }
  }'

{
  "execution_id": "exec-skill123",
  "workflow_id": "wf-skill456",
  "status": "completed",
  "result": {
    "mean": 20.36,
    "min": 10.5,
    "max": 30.2,
    "count": 5
  },
  "duration_ms": 5
}

Decorator Parameters

Prop

Type

Common Patterns

Database Operations

Deterministic data retrieval and manipulation.

from pydantic import BaseModel

class UserProfile(BaseModel):
    id: int
    name: str
    email: str
    created_at: str

@app.skill(tags=["database", "user"])
def get_user_profile(user_id: int) -> UserProfile:
    """Retrieve user profile from database."""
    user = database.query(User).filter_by(id=user_id).first()
    if not user:
        raise ValueError(f"User {user_id} not found")

    return UserProfile(
        id=user.id,
        name=user.name,
        email=user.email,
        created_at=user.created_at.isoformat()
    )

External API Integration

Async skills for external service calls.

@app.skill(tags=["api", "external"])
async def send_notification(
    user_id: int,
    message: str,
    channel: str = "email"
) -> dict:
    """Send notification via external service."""
    response = await notification_service.send(
        user_id=user_id,
        message=message,
        channel=channel
    )

    return {
        "status": "sent",
        "notification_id": response.id,
        "channel": channel,
        "timestamp": response.timestamp
    }

Data Processing

Calculations and transformations.

@app.skill(tags=["analytics", "processing"])
def process_sales_data(transactions: list[dict]) -> dict:
    """Process sales transactions and calculate metrics."""
    total_revenue = sum(t["amount"] for t in transactions)
    avg_transaction = total_revenue / len(transactions) if transactions else 0

    return {
        "total_revenue": total_revenue,
        "transaction_count": len(transactions),
        "average_transaction": avg_transaction,
        "processed_at": datetime.now().isoformat()
    }

Router-Based Organization

Group related skills with AgentRouter.

from agentfield.router import AgentRouter

app = Agent(node_id="user-service")
users = AgentRouter(prefix="Users/Management")

@users.skill(tags=["database"])
def create_user(name: str, email: str) -> dict:
    """Create new user account."""
    user = database.create_user(name=name, email=email)
    return {"user_id": user.id, "created": True}

@users.skill(tags=["database"])
def delete_user(user_id: int) -> dict:
    """Delete user account."""
    database.delete_user(user_id)
    return {"user_id": user_id, "deleted": True}

app.include_router(users)

API endpoints:

/api/v1/execute/user-service.users_management_create_user
/api/v1/execute/user-service.users_management_delete_user

Controlling Verifiable Credential Generation

Override the agent-level VC policy for specific skills based on compliance requirements.

# Force VC generation for audit-critical operations
@app.skill(tags=["database", "audit"], vc_enabled=True)
def record_financial_transaction(transaction_id: str, amount: float) -> dict:
    """Record transaction with mandatory verifiable credential."""
    database.save_transaction(transaction_id, amount)
    return {"transaction_id": transaction_id, "recorded": True}

# Disable VC for high-frequency, low-risk operations
@app.skill(tags=["cache"], vc_enabled=False)
def get_cached_data(key: str) -> dict:
    """Fast cache lookup without VC overhead."""
    return cache.get(key) or {}

# Inherit from agent-level setting (default)
@app.skill(tags=["processing"], vc_enabled=None)
def process_data(data: list) -> dict:
    """Uses agent's default VC policy."""
    return {"processed": len(data)}

VC generation hierarchy: skill decorator → agent node → platform default (enabled). Use vc_enabled=True for compliance-critical database operations, False for cache lookups and high-frequency calls, and None (default) to inherit the agent's policy.

Skills vs Reasoners

Use Skills For:

Database queries and updates
API calls to external services
Mathematical calculations
Data transformations
File operations
Deterministic business logic

Use Reasoners For:

AI-powered analysis
Natural language processing
Content generation
Decision-making with LLMs
Non-deterministic tasks

# Skill - deterministic data retrieval
@app.skill(tags=["database"])
def get_customer_orders(customer_id: int) -> list[dict]:
    """Get all orders for a customer."""
    return database.get_orders(customer_id)

# Reasoner - AI-powered analysis
@app.reasoner
async def analyze_customer_behavior(customer_id: int) -> dict:
    """Analyze customer purchase patterns with AI."""
    orders = get_customer_orders(customer_id)  # Direct call to skill

    analysis = await app.ai(
        system="You are a customer behavior analyst.",
        user=f"Analyze these orders: {orders}",
        schema=BehaviorAnalysis
    )
    return analysis

When to Use @app.skill() vs Normal Functions

The Granularity Philosophy:

Using @app.skill() is a deliberate choice about what to record in your distributed system. Each decorated skill creates:

REST API endpoint
Digital Identity (DID)
Verifiable Credential (VC)
Complete audit trail

Normal Python Function:

# Not decorated - no recording, just executes
def calculate_total(items: list[dict]) -> float:
    return sum(item["price"] for item in items)

Decorated Skill:

# Decorated - creates API, DID, VC, audit trail
@app.skill(tags=["billing"])
def calculate_total(items: list[dict]) -> float:
    return sum(item["price"] for item in items)

Choose Based on Your Needs:

More @app.skill() decorators = More granular tracking, more VCs, complete auditability
Fewer decorators = Less overhead, faster execution, but less visibility
No decorator = Just a normal function, no Agentfield integration

Use @app.skill() when you need:

API access from other agents or services
Audit trail for compliance
Verifiable credentials for the operation
Workflow tracking and monitoring

Skip the decorator when you have:

Internal helper functions
Simple calculations
Operations that don't need tracking
Performance-critical code paths

Execution Context Access

Access workflow metadata within skills.

from agentfield.execution_context import ExecutionContext

@app.skill(tags=["audit"])
def log_transaction(
    transaction_id: str,
    amount: float,
    execution_context: ExecutionContext = None
) -> dict:
    """Log transaction with execution context."""
    log_entry = {
        "transaction_id": transaction_id,
        "amount": amount,
        "timestamp": datetime.now().isoformat()
    }

    if execution_context:
        log_entry["workflow_id"] = execution_context.workflow_id
        log_entry["execution_id"] = execution_context.execution_id

    database.save_log(log_entry)
    return log_entry

The execution_context parameter is automatically injected when the skill is called via the Agentfield server.

Direct Function Calls vs app.call()

Same Agent - Direct Import

# skills/database.py
@app.skill(tags=["database"])
def get_user_data(user_id: int) -> dict:
    """Get user data from database."""
    return database.get_user(user_id)

# skills/processing.py
from skills.database import get_user_data

@app.skill(tags=["processing"])
def process_user_report(user_id: int) -> dict:
    """Process user report with direct skill call."""
    user_data = get_user_data(user_id)  # Direct call - fastest

    return {
        "user_id": user_id,
        "report": generate_report(user_data),
        "generated_at": datetime.now().isoformat()
    }

Cross-Agent - Use app.call()

@app.skill(tags=["orchestration"])
async def generate_comprehensive_report(user_id: int) -> dict:
    """Generate report using multiple agents."""

    # Call user service on different agent
    user_data = await app.call(
        "user-service.get_user_data",
        user_id=user_id
    )

    # Call analytics service on different agent
    analytics = await app.call(
        "analytics-service.calculate_metrics",
        user_id=user_id
    )

    return {
        "user": user_data,
        "analytics": analytics,
        "generated_at": datetime.now().isoformat()
    }

API Calling Patterns

Synchronous Execution

curl -X POST http://localhost:8080/api/v1/execute/data-processor.calculate_metrics \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "data": [1.5, 2.3, 3.7, 4.2]
    }
  }'

Asynchronous Execution

For long-running skills with webhook callbacks.

curl -X POST http://localhost:8080/api/v1/execute/async/data-processor.process_large_dataset \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "dataset_id": "dataset_12345"
    },
    "webhook": {
      "url": "https://your-app.com/agentfield/callback",
      "secret": "your-webhook-secret"
    }
  }'

Error Handling

Skills should handle errors explicitly.

@app.skill(tags=["database"])
def get_user_safe(user_id: int) -> dict:
    """Get user with error handling."""
    try:
        user = database.get_user(user_id)
        if not user:
            return {"error": "User not found", "user_id": user_id}

        return {
            "success": True,
            "user": {
                "id": user.id,
                "name": user.name,
                "email": user.email
            }
        }
    except DatabaseError as e:
        return {
            "error": "Database error",
            "message": str(e),
            "user_id": user_id
        }

Performance Considerations

Skills are Lightweight:

No AI/LLM overhead
No workflow tracking overhead (unless called via app.call())
Direct function execution
Ideal for high-frequency operations

Optimization Tips:

# Fast - direct function call
from skills.utils import calculate_total

@app.skill
def process_order(items: list[dict]) -> dict:
    total = calculate_total(items)  # Direct call, no network
    return {"total": total, "items": len(items)}

# Slower - cross-agent call (but enables distributed architecture)
@app.skill
async def process_distributed_order(items: list[dict]) -> dict:
    total = await app.call(
        "calculator-service.calculate_total",  # Network call
        items=items
    )
    return {"total": total, "items": len(items)}

Agent Node - Core agent initialization and configuration
@app.reasoner() - AI-powered functions
app.call() - Cross-agent communication
app.memory - Shared state management
Agent Router - Organize skills with routers
Make Your Agent Async - Async execution patterns