Project structure

# Scaffold a production project in the current directory without prompts
10xgraph init --template production --auth jwt --rate-limit redis --name MyAgent --yes

from datetime import datetime

from tenxgraph.core.state import MessageContextManager
from tenxgraph.prebuilt.agent import ReactAgent
from tenxgraph.storage.checkpointer import InMemoryCheckpointer
from dotenv import load_dotenv

from graph.state import WeatherState
from graph.tools.weather_tool import get_weather
from graph.validators.manager import callback_manager


load_dotenv()

checkpointer = InMemoryCheckpointer()

context_manager = MessageContextManager(
    max_messages=20,  # last 20 user messages will be kept in context
    remove_tool_msgs=True,
)

react_agent = ReactAgent(
    state=WeatherState(),
    model="google/gemini-2.5-flash",
    provider="google",
    system_prompt=[
        {
            "role": "system",
            "content": """
                You are a helpful assistant.
                Your task is to assist the user in finding information and answering questions.
                User Current Location: {user_location}
                If missing information, ask the user for clarification.
                Use tools when they help answer the user.
            """,
        },
        {
            "role": "user",
            "content": f"Today Date is {datetime.now().strftime('%Y-%m-%d')}",
        },
    ],
    trim_context=True,
    tools=[get_weather],
    context_manager=context_manager,
)

app = react_agent.compile(
    checkpointer=checkpointer,
    callback_manager=callback_manager,
)

from tenxgraph.core import AgentState


class WeatherState(AgentState):
    user_location: str = ""

import logging
import random

from tenxgraph.core.state.stream_emitter import StreamEmitter

from graph.state import WeatherState


logger = logging.getLogger(__name__)

MAX_RETRIES = 3


def call_weather_api(location: str) -> str:
    is_failed = random.choice([True, False])  # noqa: S311
    if is_failed:
        raise Exception("Failed to fetch weather data due to a simulated API error.")
    return f"The weather in {location} is sunny"


def get_weather(
    location: str,
    tool_call_id: str | None = None,
    state: WeatherState | None = None,
    emit: StreamEmitter | None = None,
) -> str:
    """Get the current weather for a specific location."""
    if tool_call_id:
        logger.info(f"Tool call ID: {tool_call_id}")
    if state and hasattr(state, "context"):
        logger.info(f"Number of messages in context: {len(state.context)}")  # type: ignore
    if emit:
        emit.progress("Fetching weather data...")

    for i in range(MAX_RETRIES):
        try:
            return call_weather_api(location)
        except Exception as e:
            logger.error(f"Attempt {i + 1} failed: {e}")
            if emit:
                emit.progress(f"Attempt {i + 1} failed, retrying...")

    return f"Sorry, I couldn't fetch the weather for {location} after multiple attempts."

from tenxgraph.utils.validators import PromptInjectionValidator


prompt_validator = PromptInjectionValidator(
    strict_mode=True,
    max_length=1000,
    blocked_patterns=[],
    suspicious_keywords=[
        "ignore previous",
        "forget previous",
        "disregard previous",
        "bypass",
        "circumvent",
        "override",
        "disable",
        "remove restrictions",
        "token",
        "coupon",
        "free",
    ],
)

from tenxgraph.utils.callbacks import CallbackManager

from .lifecyle import AgentLifecycleHook
from .validators import prompt_validator


callback_manager = CallbackManager()
callback_manager.register_input_validator(prompt_validator)
callback_manager.register_lifecycle_hook(AgentLifecycleHook())

from tenxgraph_api import ThreadNameGenerator


class MyNameGenerator(ThreadNameGenerator):
    async def generate_name(self, messages: list[str]) -> str:
        # TODO: Implement logic to generate thread name based on messages
        return "MyCustomThreadName"

import logging
from typing import Any

from tenxgraph_api import BaseAuth
from fastapi import HTTPException, Request, Response
from fastapi.security import HTTPAuthorizationCredentials


logger = logging.getLogger(__name__)


class AgentAuth(BaseAuth):
    """Verify the caller and return who they are."""

    def __init__(self) -> None:
        logger.warning(
            "AgentAuth.authenticate is not implemented yet: every request will get 401. "
            "Implement it in auth/agent_auth.py."
        )

    def authenticate(
        self,
        request: Request,
        response: Response,
        credential: HTTPAuthorizationCredentials,
    ) -> dict[str, Any] | None:
        if credential is None:
            raise HTTPException(status_code=401, detail="Missing credentials")

        # Replace this with a real check, for example verifying the bearer token with your
        # identity provider:
        #
        #     claims = verify_token(credential.credentials)  # raises on a bad token
        #     return {"user_id": claims["sub"], "roles": claims.get("roles", [])}
        raise HTTPException(status_code=401, detail="Authentication is not configured")

APP_NAME="MyAgent"
APP_VERSION="0.1.0"
MODE="production"
LOG_LEVEL="INFO"
IS_DEBUG="false"
SUMMARY="MyAgent Backend"

ORIGINS="*"
ALLOWED_HOST="*"

ROOT_PATH="/"
DOCS_PATH=""
REDOCS_PATH=""

MAX_REQUEST_SIZE=10485760  # 10MB

SECURITY_HEADERS_ENABLED="true"
HSTS_ENABLED="true"

SNOWFLAKE_EPOCH=1609459200000
SNOWFLAKE_NODE_ID=1
SNOWFLAKE_WORKER_ID=2

SENTRY_DSN=""

# Only with --rate-limit redis
REDIS_URL=""

# Only with --auth jwt
JWT_SECRET_KEY=""
JWT_ALGORITHM="HS256"

GOOGLE_API_KEY=""

{
  "agent": "graph.agent:app",
  "env": ".env",
  "auth": {
    "method": "custom",
    "path": "auth.agent_auth:AgentAuth"
  },
  "thread_name_generator": "graph.thread_name_generator:MyNameGenerator",
  "ag_ui": {
    "enabled": false
  },
  "authorization": "ownership",
  "injectq": "graph.agent:container",
  "rate_limit": {
    "enabled": true,
    "backend": "redis",
    "requests": 100,
    "window": 60,
    "by": "ip",
    "trusted_proxy_headers": false,
    "exclude_paths": ["/ping", "/docs", "/redoc", "/openapi.json"]
  }
}