AI Agents in 2027: What They Can Actually Do, Best Tools, Real Business Use Cases & How to Build Your Own

from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class AgentState(TypedDict):
    topic: str
    search_queries: List[str]
    raw_sources: List[dict]
    summary: str
    is_verified: bool

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def plan_queries(state: AgentState):
    prompt = f"Generate 3 distinct search queries to thoroughly research: {state['topic']}"
    res = llm.invoke(prompt)
    queries = [q.strip() for q in res.content.split("\n") if q.strip()][:3]
    return {"search_queries": queries}

def search_web(state: AgentState):
    # Simulated search tool call (replace with Exa or Brave API)
    sources = [{"url": "https://nist.gov", "text": "NIST released FIPS 203 in August 2024."}]
    return {"raw_sources": sources}

def synthesize_and_verify(state: AgentState):
    prompt = f"Synthesize this research topic: {state['topic']} based strictly on sources: {state['raw_sources']}"
    res = llm.invoke(prompt)
    # Critic verify: Ensure claims are present in raw sources
    verified = "NIST" in res.content and "FIPS 203" in res.content
    return {"summary": res.content, "is_verified": verified}

# Build StateGraph
workflow = StateGraph(AgentState)
workflow.add_node("planner", plan_queries)
workflow.add_node("searcher", search_web)
workflow.add_node("synthesizer", synthesize_and_verify)

workflow.set_entry_point("planner")
workflow.add_edge("planner", "searcher")
workflow.add_edge("searcher", "synthesizer")
workflow.add_edge("synthesizer", END)

app = workflow.compile()
# Run agent
result = app.invoke({"topic": "NIST Post-Quantum Timeline"})
print(result["summary"])

import { google } from 'googleapis';
import Anthropic from '@anthropic-ai/sdk';

const anthropic = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY });
const oauth2Client = new google.auth.OAuth2(
  process.env.GOOGLE_CLIENT_ID,
  process.env.GOOGLE_CLIENT_SECRET,
  process.env.GOOGLE_REDIRECT_URI
);
oauth2Client.setCredentials({ refresh_token: process.env.GOOGLE_REFRESH_TOKEN });
const gmail = google.gmail({ version: 'v1', auth: oauth2Client });

export async function processInboundEmail(sender: string, subject: string, body: string) {
  // Step 1: Prompt Injection Defense Linter
  if (body.toLowerCase().includes("ignore previous instructions")) {
    console.warn("Security Alert: Malicious prompt injection detected from", sender);
    return { status: "quarantined" };
  }

  // Step 2: Generate Draft with Claude 3.5 Haiku
  const response = await anthropic.messages.create({
    model: "claude-3-5-haiku-20241022",
    max_tokens: 500,
    system: "You are an executive email assistant for Tenzed Technologies. Draft polite, concise replies. Never make financial promises.",
    messages: [{ role: "user", content: `Sender: ${sender}\nSubject: ${subject}\nBody: ${body}\nDraft a professional response.` }]
  });

  const draftText = response.content[0].type === 'text' ? response.content[0].text : '';

  // Step 3: Create Draft in Gmail (NEVER SEND DIRECTLY)
  const rawEmail = [
    `To: ${sender}`,
    'Content-Type: text/plain; charset=utf-8',
    'MIME-Version: 1.0',
    `Subject: Re: ${subject}`,
    '',
    draftText
  ].join('\n');

  const encodedDraft = Buffer.from(rawEmail).toString('base64url');

  await gmail.users.drafts.create({
    userId: 'me',
    requestBody: { message: { raw: encodedDraft } }
  });

  console.log("✓ Safe draft created successfully for human review:", subject);
  return { status: "draft_created", draftText };
}

from pydantic import BaseModel, EmailStr
from openai import OpenAI
import requests
import os

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

class LeadSubmission(BaseModel):
    name: str
    email: EmailStr
    company: str
    requirement: str
    budget_range: str

class QualificationResult(BaseModel):
    score: int
    is_qualified: bool
    reasoning: str
    deal_priority: str

def qualify_lead(lead: LeadSubmission) -> QualificationResult:
    # Reject disposable / free mailboxes
    free_domains = ["gmail.com", "yahoo.com", "hotmail.com"]
    domain = lead.email.split("@")[1].lower()
    if domain in free_domains:
        return QualificationResult(score=15, is_qualified=False, reasoning="Free email domain", deal_priority="LOW")

    # Evaluate via Structured Outputs
    prompt = f"""Evaluate this B2B lead for enterprise software development:
Company: {lead.company}
Requirement: {lead.requirement}
Budget: {lead.budget_range}
Score 1-100 based on BANT criteria. High priority if budget > $20k and urgency is high."""

    completion = client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[{"role": "user", "content": prompt}],
        response_format=QualificationResult,
    )
    
    result = completion.choices[0].message.parsed
    # If qualified, sync to HubSpot CRM
    if result.is_qualified:
        headers = {"Authorization": f"Bearer {os.getenv('HUBSPOT_API_KEY')}", "Content-Type": "application/json"}
        payload = {
            "properties": {
                "email": lead.email,
                "firstname": lead.name,
                "company": lead.company,
                "lead_score": str(result.score),
                "hs_lead_status": "QUALIFIED"
            }
        }
        requests.post("https://api.hubapi.com/crm/v3/objects/contacts", headers=headers, json=payload)
    
    return result

import { OpenAI } from 'openai';
import { z } from 'zod';

const openai = new OpenAI();

// Define Tool Schema with strict validation
const LookupOrderSchema = z.object({
  orderId: z.string().regex(/^[0-9A-Z_-]+$/, "Invalid order ID format")
});

const tools: OpenAI.ChatCompletionTool[] = [
  {
    type: "function",
    function: {
      name: "lookup_order",
      description: "Lookup real-time shipping and fulfillment status for a customer order.",
      parameters: {
        type: "object",
        properties: {
          orderId: { type: "string", description: "The unique order identifier, e.g. ORD-4891" }
        },
        required: ["orderId"],
        additionalProperties: false
      }
    }
  }
];

// Execution handler on READ-ONLY database replica
async function executeLookupOrder(orderId: string, authenticatedUserId: string) {
  // Security verification: Ensure order belongs to caller
  console.log(`[DB Audit] SELECT status FROM orders WHERE id = ${orderId} AND user_id = ${authenticatedUserId}`);
  return JSON.stringify({ orderId, status: "IN_TRANSIT", carrier: "FedEx", estDelivery: "Tomorrow by 5 PM" });
}

export async function runSupportAgent(userMessage: string, userId: string) {
  const runner = await openai.chat.completions.create({
    model: "gpt-4o",
    messages: [
      { role: "system", content: "You are a customer support agent. Use tools to check order status. Never guess shipping dates." },
      { role: "user", content: userMessage }
    ],
    tools: tools,
    tool_choice: "auto"
  });

  const toolCall = runner.choices[0].message.tool_calls?.[0];
  if (toolCall && toolCall.function.name === "lookup_order") {
    const args = LookupOrderSchema.parse(JSON.parse(toolCall.function.arguments));
    const toolResult = await executeLookupOrder(args.orderId, userId);

    const finalResponse = await openai.chat.completions.create({
      model: "gpt-4o",
      messages: [
        { role: "system", content: "Explain the order status clearly to the customer." },
        { role: "user", content: userMessage },
        runner.choices[0].message,
        { role: "tool", tool_call_id: toolCall.id, content: toolResult }
      ]
    });
    return finalResponse.choices[0].message.content;
  }
  return runner.choices[0].message.content;
}

from pydantic import BaseModel, Field
from typing import List
from openai import OpenAI
import os

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

class InvoiceLineItem(BaseModel):
    description: str
    quantity: float
    unit_price: float
    line_total: float

class ExtractedInvoice(BaseModel):
    vendor_name: str
    invoice_number: str
    invoice_date: str
    line_items: List[InvoiceLineItem]
    subtotal: float
    tax_amount: float
    total_amount: float

def parse_invoice_text(invoice_raw_text: str) -> dict:
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[
            {"role": "system", "content": "Extract structured invoice data with 100% precision. Never round numbers."},
            {"role": "user", "content": f"Invoice Text:\n{invoice_raw_text}"}
        ],
        response_format=ExtractedInvoice,
    )
    
    invoice = completion.choices[0].message.parsed
    
    # Mathematical reconciliation verification in Python
    calculated_subtotal = sum(item.quantity * item.unit_price for item in invoice.line_items)
    is_valid_math = abs(calculated_subtotal - invoice.subtotal) < 0.05
    
    return {
        "data": invoice.model_dump(),
        "is_mathematically_valid": is_valid_math,
        "requires_human_review": not is_valid_math
    }

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class MultiAgentState(TypedDict):
    task: str
    code: str
    feedback: str
    iterations: int
    is_approved: bool

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def coder_agent(state: MultiAgentState):
    prompt = f"Implement task: {state['task']}. Previous feedback: {state.get('feedback', 'None')}"
    res = llm.invoke(prompt)
    return {
        "code": res.content,
        "iterations": state.get("iterations", 0) + 1
    }

def reviewer_agent(state: MultiAgentState):
    prompt = f"Review this code for bugs and concurrency flaws:\n{state['code']}\nIf perfect, reply 'APPROVED'. Otherwise explain flaws."
    res = llm.invoke(prompt)
    approved = "APPROVED" in res.content
    return {
        "feedback": res.content,
        "is_approved": approved
    }

def router(state: MultiAgentState) -> Literal["coder", "__end__"]:
    if state.get("is_approved") or state.get("iterations", 0) >= 3:
        return "__end__"
    return "coder"

# Construct Orchestration Graph
workflow = StateGraph(MultiAgentState)
workflow.add_node("coder", coder_agent)
workflow.add_node("reviewer", reviewer_agent)

workflow.set_entry_point("coder")
workflow.add_edge("coder", "reviewer")
workflow.add_conditional_edges("reviewer", router, {"coder": "coder", "__end__": END})

app = workflow.compile()
print("✓ Multi-Agent Mesh compiled and ready for execution.")