Pydantic AI

pip install -U pydantic-ai

import os
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

model = OpenAIChatModel(
    "deepseek-v4.1-flash",
    provider=OpenAIProvider(
        base_url="https://api.neosantara.xyz/v1",
        api_key=os.environ["NEOSANTARA_API_KEY"],
    ),
)

agent = Agent(
    model,
    system_prompt="You are a concise technical assistant."
)

result = agent.run_sync("What is the primary role of an API gateway rate limiter?")
print(result.data)

import os
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

class ServerMetricReport(BaseModel):
    server_id: str = Field(description="Unique node identifier")
    healthy: bool = Field(description="Operational health flag")
    cpu_usage_pct: float = Field(description="CPU load percentage")
    summary: str = Field(description="1-sentence diagnosis")

model = OpenAIChatModel(
    "gemini-3.8-flash",
    provider=OpenAIProvider(
        base_url="https://api.neosantara.xyz/v1",
        api_key=os.environ["NEOSANTARA_API_KEY"],
    ),
)

agent = Agent(
    model,
    result_type=ServerMetricReport,
    system_prompt="Extract telemetry metrics from raw server logs into the structured format."
)

log_text = "Node jkt-prod-02 is operating normally at 42C with recorded CPU utilization of 28.4%."
result = agent.run_sync(log_text)

print(f"Server: {result.data.server_id}")
print(f"Healthy: {result.data.healthy}")
print(f"CPU Load: {result.data.cpu_usage_pct}%")
print(f"Summary: {result.data.summary}")

import os
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

model = OpenAIChatModel(
    "deepseek-v4.1-flash",
    provider=OpenAIProvider(
        base_url="https://api.neosantara.xyz/v1",
        api_key=os.environ["NEOSANTARA_API_KEY"],
    ),
)

agent = Agent(model, system_prompt="Assist developers with token cost estimations.")

@agent.tool
def calculate_cost_idr(ctx: RunContext[None], token_count: int, rate_per_million: float) -> str:
    """Calculate the total Rupiah cost for a given token volume."""
    total = (token_count / 1_000_000) * rate_per_million
    return f"Rp {total:,.2f}"

result = agent.run_sync("What is the cost for 2,500,000 output tokens if the rate is Rp 4,500 per million tokens?")
print(result.data)