personamanagmentlayer__pcl/stdlib/api/microservices-expert/SKILL.md

name, version, description, category, tags, allowed-tools

name	version	description	category	tags	allowed-tools
microservices-expert	1.1.0	Expert-level microservices architecture, patterns, service mesh, and distributed systems. Use when the user mentions distributed systems, service mesh, or architecture, or when the task involves Microservices Principles, Architecture Patterns, Communication, or Data Management.	api	microservices distributed-systems service-mesh architecture	Read Write Edit

Microservices Expert

Expert guidance for microservices architecture, design patterns, service communication, and distributed system challenges.

Core Concepts

Microservices Principles

• Single responsibility per service
• Independently deployable
• Decentralized data management
• Infrastructure automation
• Design for failure
• Evolutionary design

Architecture Patterns

• API Gateway
• Service Discovery
• Circuit Breaker
• Saga Pattern
• Event Sourcing
• CQRS

Communication

• Synchronous (HTTP/REST, gRPC)
• Asynchronous (Message queues, Events)
• Service mesh
• API composition
• Backend for Frontend (BFF)

Service Design

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx
from typing import List, Optional
from circuitbreaker import circuit
import asyncio

# Individual Microservice
app = FastAPI(title="Order Service", version="1.0.0")

class Order(BaseModel):
    id: str
    user_id: str
    items: List[dict]
    total: float
    status: str

class OrderService:
    def __init__(self, inventory_url: str, payment_url: str):
        self.inventory_url = inventory_url
        self.payment_url = payment_url
        self.client = httpx.AsyncClient()

    @circuit(failure_threshold=5, recovery_timeout=60)
    async def check_inventory(self, items: List[dict]) -> bool:
        """Check inventory availability with circuit breaker"""
        try:
            response = await self.client.post(
                f"{self.inventory_url}/check",
                json={"items": items},
                timeout=5.0
            )
            return response.json()["available"]
        except Exception as e:
            print(f"Inventory service error: {e}")
            raise

    @circuit(failure_threshold=5, recovery_timeout=60)
    async def process_payment(self, user_id: str, amount: float) -> dict:
        """Process payment with circuit breaker"""
        try:
            response = await self.client.post(
                f"{self.payment_url}/charge",
                json={"user_id": user_id, "amount": amount},
                timeout=10.0
            )
            return response.json()
        except Exception as e:
            print(f"Payment service error: {e}")
            raise

    async def create_order(self, order: Order) -> Order:
        """Create order with coordination"""
        # 1. Check inventory
        inventory_available = await self.check_inventory(order.items)
        if not inventory_available:
            raise HTTPException(400, "Items not available")

        # 2. Process payment
        payment = await self.process_payment(order.user_id, order.total)
        if payment["status"] != "success":
            raise HTTPException(400, "Payment failed")

        # 3. Reserve inventory
        await self.reserve_inventory(order.items)

        # 4. Create order record
        order.status = "confirmed"
        await self.save_order(order)

        return order

@app.post("/orders", response_model=Order)
async def create_order(order: Order):
    service = OrderService(
        inventory_url="http://inventory-service",
        payment_url="http://payment-service"
    )
    return await service.create_order(order)

Saga Pattern (Distributed Transactions)

from enum import Enum
from typing import List, Callable
import asyncio

class SagaStep:
    def __init__(self, action: Callable, compensation: Callable):
        self.action = action
        self.compensation = compensation

class SagaOrchestrator:
    """Orchestrate distributed transactions using Saga pattern"""

    def __init__(self):
        self.steps: List[SagaStep] = []
        self.completed_steps: List[SagaStep] = []

    def add_step(self, action: Callable, compensation: Callable):
        """Add a step to the saga"""
        self.steps.append(SagaStep(action, compensation))

    async def execute(self) -> bool:
        """Execute saga with compensation on failure"""
        try:
            # Execute all steps
            for step in self.steps:
                result = await step.action()
                self.completed_steps.append(step)

                if not result:
                    await self.compensate()
                    return False

            return True

        except Exception as e:
            print(f"Saga failed: {e}")
            await self.compensate()
            return False

    async def compensate(self):
        """Rollback completed steps"""
        # Execute compensations in reverse order
        for step in reversed(self.completed_steps):
            try:
                await step.compensation()
            except Exception as e:
                print(f"Compensation failed: {e}")

# Example: Order Saga
class OrderSaga:
    async def create_order_with_saga(self, order_data: dict):
        saga = SagaOrchestrator()

        # Step 1: Reserve inventory
        saga.add_step(
            action=lambda: self.reserve_inventory(order_data["items"]),
            compensation=lambda: self.release_inventory(order_data["items"])
        )

        # Step 2: Process payment
        saga.add_step(
            action=lambda: self.charge_payment(order_data["user_id"], order_data["total"]),
            compensation=lambda: self.refund_payment(order_data["user_id"], order_data["total"])
        )

        # Step 3: Create order
        saga.add_step(
            action=lambda: self.create_order_record(order_data),
            compensation=lambda: self.delete_order_record(order_data["id"])
        )

        # Execute saga
        success = await saga.execute()

        if success:
            await self.send_confirmation(order_data["user_id"])
            return {"status": "success", "order_id": order_data["id"]}
        else:
            return {"status": "failed", "message": "Order creation failed"}

Service Discovery

import consul
from typing import List, Optional
import random

class ServiceRegistry:
    """Service discovery using Consul"""

    def __init__(self, consul_host: str = "localhost", consul_port: int = 8500):
        self.consul = consul.Consul(host=consul_host, port=consul_port)

    def register_service(self, service_name: str, service_id: str,
                        host: str, port: int, tags: List[str] = None):
        """Register service with Consul"""
        self.consul.agent.service.register(
            name=service_name,
            service_id=service_id,
            address=host,
            port=port,
            tags=tags or [],
            check=consul.Check.http(
                f"http://{host}:{port}/health",
                interval="10s",
                timeout="5s"
            )
        )

    def deregister_service(self, service_id: str):
        """Deregister service"""
        self.consul.agent.service.deregister(service_id)

    def discover_service(self, service_name: str) -> Optional[dict]:
        """Discover healthy service instance"""
        _, services = self.consul.health.service(service_name, passing=True)

        if not services:
            return None

        # Load balance: random selection
        service = random.choice(services)

        return {
            "id": service["Service"]["ID"],
            "address": service["Service"]["Address"],
            "port": service["Service"]["Port"],
            "tags": service["Service"]["Tags"]
        }

    def get_all_services(self, service_name: str) -> List[dict]:
        """Get all healthy instances of a service"""
        _, services = self.consul.health.service(service_name, passing=True)

        return [
            {
                "id": s["Service"]["ID"],
                "address": s["Service"]["Address"],
                "port": s["Service"]["Port"]
            }
            for s in services
        ]

API Gateway

from fastapi import FastAPI, Request, Response
import httpx
from typing import Dict
import jwt

app = FastAPI(title="API Gateway")

class APIGateway:
    """API Gateway for routing and cross-cutting concerns"""

    def __init__(self):
        self.service_registry = ServiceRegistry()
        self.client = httpx.AsyncClient()

    async def route_request(self, service: str, path: str,
                           method: str, **kwargs) -> Response:
        """Route request to appropriate microservice"""
        # Discover service
        service_info = self.service_registry.discover_service(service)

        if not service_info:
            return Response(
                content={"error": "Service unavailable"},
                status_code=503
            )

        # Build URL
        url = f"http://{service_info['address']}:{service_info['port']}{path}"

        # Forward request
        response = await self.client.request(method, url, **kwargs)

        return Response(
            content=response.content,
            status_code=response.status_code,
            headers=dict(response.headers)
        )

    async def authenticate(self, token: str) -> Optional[dict]:
        """Centralized authentication"""
        try:
            payload = jwt.decode(token, "secret", algorithms=["HS256"])
            return payload
        except jwt.JWTError:
            return None

    async def rate_limit(self, client_id: str) -> bool:
        """Centralized rate limiting"""
        # Implementation using Redis
        pass

# Gateway endpoints
@app.api_route("/{service}/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def gateway_route(service: str, path: str, request: Request):
    gateway = APIGateway()

    # Authentication
    token = request.headers.get("Authorization", "").replace("Bearer ", "")
    user = await gateway.authenticate(token)

    if not user:
        return Response(content={"error": "Unauthorized"}, status_code=401)

    # Rate limiting
    if not await gateway.rate_limit(user["id"]):
        return Response(content={"error": "Rate limit exceeded"}, status_code=429)

    # Route request
    return await gateway.route_request(
        service=service,
        path=f"/{path}",
        method=request.method,
        headers=dict(request.headers),
        content=await request.body()
    )

Event-Driven Architecture

import pika
import json
from typing import Callable, Dict
import asyncio

class EventBus:
    """Message broker for event-driven communication"""

    def __init__(self, rabbitmq_url: str):
        self.connection = pika.BlockingConnection(
            pika.URLParameters(rabbitmq_url)
        )
        self.channel = self.connection.channel()
        self.handlers: Dict[str, Callable] = {}

    def publish_event(self, event_type: str, data: dict):
        """Publish event to all subscribers"""
        self.channel.exchange_declare(
            exchange='events',
            exchange_type='topic',
            durable=True
        )

        message = json.dumps({
            "event_type": event_type,
            "data": data,
            "timestamp": datetime.now().isoformat()
        })

        self.channel.basic_publish(
            exchange='events',
            routing_key=event_type,
            body=message,
            properties=pika.BasicProperties(
                delivery_mode=2  # persistent
            )
        )

    def subscribe(self, event_type: str, handler: Callable):
        """Subscribe to event type"""
        self.handlers[event_type] = handler

        # Declare queue
        queue_name = f"{event_type}_queue"
        self.channel.queue_declare(queue=queue_name, durable=True)

        # Bind queue to exchange
        self.channel.queue_bind(
            queue=queue_name,
            exchange='events',
            routing_key=event_type
        )

        # Start consuming
        def callback(ch, method, properties, body):
            message = json.loads(body)
            handler(message["data"])
            ch.basic_ack(delivery_tag=method.delivery_tag)

        self.channel.basic_consume(
            queue=queue_name,
            on_message_callback=callback
        )

    def start_consuming(self):
        """Start consuming events"""
        self.channel.start_consuming()

# Usage Example
event_bus = EventBus("amqp://localhost")

# Service A publishes event
event_bus.publish_event("order.created", {
    "order_id": "12345",
    "user_id": "user_1",
    "total": 99.99
})

# Service B subscribes to event
def handle_order_created(data):
    print(f"Processing order: {data['order_id']}")
    # Send email, update inventory, etc.

event_bus.subscribe("order.created", handle_order_created)

Best Practices

Design

• Keep services small and focused
• Design for failure (circuit breakers, retries)
• Use asynchronous communication when possible
• Implement proper service boundaries
• Avoid distributed monoliths
• Use API versioning
• Implement health checks

Data Management

• Database per service
• Use eventual consistency
• Implement saga pattern for distributed transactions
• Use event sourcing for audit trails
• Cache aggressively
• Avoid distributed joins

Operations

• Implement distributed tracing
• Centralized logging
• Monitor service health
• Automate deployments
• Use service mesh for cross-cutting concerns
• Implement feature flags
• Practice chaos engineering

Anti-Patterns

❌ Distributed monolith ❌ Shared database between services ❌ Synchronous communication everywhere ❌ No service versioning ❌ Tight coupling between services ❌ No circuit breakers ❌ Missing distributed tracing

Resources

• Microservices Patterns: https://microservices.io/patterns/
• Building Microservices (book)
• Service Mesh: https://istio.io/
• Consul: https://www.consul.io/
• RabbitMQ: https://www.rabbitmq.com/