Core Concepts

# Your Python agent stays exactly as it is
def my_agent(query: str) -> str:
    return langgraph_agent.invoke(query)

// Your JavaScript team uses it natively
const result = await client.run({ query: "Hello!" });

// Your Rust team gets the same experience
let result = client.run(&[("query", json!("Hello!"))]).await?;

// Run the Go agent
result, err := agentClient.Run(ctx, map[string]interface{}{
	"query": "Hello!",
	})

def generate_story(prompt: str, style: str) -> Iterator[str]:
    """Stream a story as it's being generated"""
    for chunk in llm.stream(f"Write a {style} story about: {prompt}"):
        yield chunk.content

{
  "agent_name": "problem-solver",
  "description": "Solves complex problems with reasoning",
  "framework": "langgraph",
  "version": "1.0.0",
  "agent_architecture": {
    "entrypoints": [
      {
        "file": "agent.py",
        "module": "solve_problem",
        "tag": "solve"
      },
      {
        "file": "agent.py", 
        "module": "generate_story",
        "tag": "story_stream"
      }
    ]
  },
  "env_vars": {
    "OPENAI_API_KEY": "${OPENAI_API_KEY}",
    "CUSTOM_SETTING": "production"
  }
}

runagent serve .

runagent deploy .  # Coming soon

def chat_agent(message: str, conversation_id: str = None) -> dict:
    """Stateful conversation agent"""
    # Load conversation history
    history = get_conversation(conversation_id) if conversation_id else []
    
    # Add user message
    history.append({"role": "user", "content": message})
    
    # Generate response
    response = llm.chat(history)
    
    # Save updated history
    save_conversation(conversation_id, history + [response])
    
    return {
        "message": response.content,
        "conversation_id": conversation_id or generate_id()
    }

def tool_agent(query: str, available_tools: list = None) -> dict:
    """Agent that can use external tools"""
    tools = load_tools(available_tools or ["search", "calculator"])
    
    # Let the agent decide which tools to use
    plan = planner.create_plan(query, tools)
    
    results = []
    for step in plan.steps:
        if step.tool:
            result = tools[step.tool].execute(step.params)
            results.append(result)
    
    # Synthesize final response
    final_response = synthesizer.combine(query, results)
    
    return {
        "response": final_response,
        "tools_used": [step.tool for step in plan.steps if step.tool],
        "intermediate_results": results
    }

def analyze_document(document: str, analysis_type: str) -> Iterator[str]:
    """Stream analysis results as they're generated"""
    yield f"🔍 Starting {analysis_type} analysis...\n\n"
    
    # Break document into chunks
    chunks = chunk_document(document)
    
    for i, chunk in enumerate(chunks):
        yield f"📄 Analyzing section {i+1}/{len(chunks)}...\n"
        
        analysis = analyzer.analyze(chunk, analysis_type)
        yield f"**Section {i+1} Summary:**\n{analysis}\n\n"
    
    yield "✅ Analysis complete!\n"

# ✅ Excellent: Load models at module level
from transformers import pipeline

# This runs once when the module is imported
sentiment_analyzer = pipeline("sentiment-analysis")
search_client = SearchClient(api_key=os.getenv("SEARCH_API_KEY"))

def analyze_sentiment(text: str) -> dict:
    """This entrypoint reuses pre-loaded resources"""
    result = sentiment_analyzer(text)
    return {
        "sentiment": result[0]["label"],
        "confidence": result[0]["score"]
    }

# ❌ Terrible: Loading models repeatedly
def analyze_sentiment_bad(text: str) -> dict:
    # This runs on EVERY request - horrible performance!
    analyzer = pipeline("sentiment-analysis")  
    result = analyzer(text)
    return {"sentiment": result[0]["label"]}

def robust_agent(query: str) -> dict:
    """Error handling that works across all languages"""
    try:
        # Validate inputs early
        if not query or len(query.strip()) == 0:
            return {
                "success": False,
                "error_code": "EMPTY_QUERY",
                "message": "Query cannot be empty",
                "retry_possible": True,
                "suggested_action": "Provide a non-empty query string"
            }
        
        result = process_query(query)
        return {
            "success": True,
            "data": result,
            "processing_time": timer.elapsed()
        }
        
    except ValidationError as e:
        return {
            "success": False,
            "error_code": "VALIDATION_ERROR",
            "message": f"Invalid input: {str(e)}",
            "retry_possible": True,
            "invalid_fields": e.fields if hasattr(e, 'fields') else []
        }
    except RateLimitError as e:
        return {
            "success": False,
            "error_code": "RATE_LIMITED",
            "message": "Too many requests",
            "retry_possible": True,
            "retry_after_seconds": e.retry_after
        }
    except Exception as e:
        logger.error(f"Unexpected error in robust_agent: {e}", exc_info=True)
        return {
            "success": False,
            "error_code": "INTERNAL_ERROR", 
            "message": "An unexpected error occurred",
            "retry_possible": False,
            "request_id": generate_request_id()
        }

# ✅ Clear, self-documenting parameters
def search_and_summarize(
    query: str,
    max_results: int = 10,
    summary_length: str = "medium"  # "short", "medium", "long"
) -> dict:
    return {
        "summary": "...",
        "sources": [...],
        "total_results_found": 150,
        "results_used": 10
    }

# ✅ Rich return types with metadata
def process_document(document_content: str, options: dict = None) -> dict:
    return {
        "processed_content": "...",
        "statistics": {
            "word_count": 1500,
            "reading_time_minutes": 6,
            "complexity_score": 0.7
        },
        "metadata": {
            "processing_version": "2.1.0",
            "timestamp": datetime.utcnow().isoformat()
        }
    }