Agent

from logicore.agents.agent import Agent

Agent(
    llm: str | LLMProvider = "ollama",
    model: str = None,
    api_key: str = None,
    endpoint: str = None,
    system_message: str = None,
    role: str = "general",
    debug: bool = False,
    tools: list = [],
    max_iterations: int = 40,
    capabilities: Any = None,
    telemetry: bool = False,
    memory: bool = False,
    context_compression: bool = False,
    skills: list = None,
    workspace_root: str = None,
)

await agent.chat(
    user_input: str | list,
    session_id: str = "default",
    callbacks: dict = None,
    stream: bool = False,
    streaming_funct: callable = None,
    generate_walkthrough: bool = False,
    **kwargs,
) -> str

def check_stock(ticker: str, exchange: str = "NASDAQ") -> dict:
    """
    Fetch current stock price.

    Args:
        ticker: Stock ticker symbol (e.g. AAPL).
        exchange: Exchange name.
    """
    return {"ticker": ticker, "price": 182.50}

agent = Agent(llm="ollama")
agent.register_tool_from_function(check_stock)
agent.set_auto_approve_all(True)

response = await agent.chat("What is Apple's current stock price?")

schema = {
    "type": "function",
    "function": {
        "name": "send_email",
        "description": "Send an email to a recipient.",
        "parameters": {
            "type": "object",
            "properties": {
                "to": {"type": "string", "description": "Recipient email address"},
                "subject": {"type": "string", "description": "Email subject line"},
                "body": {"type": "string", "description": "Email body text"},
            },
            "required": ["to", "subject", "body"],
        },
    },
}

async def send_email(to: str, subject: str, body: str):
    # your implementation
    return {"status": "sent", "to": to}

agent.add_custom_tool(schema, send_email)

from logicore.skills import Skill

# Load by name (resolved from defaults directory or workspace)
agent.load_skills(["web_research", "code_review"])

# Load a Skill object directly
custom_skill = Skill(
    name="data_analysis",
    description="Analyze tabular data.",
    instructions="Use the provided tools to analyze CSV or JSON data.",
    tools=[...],
    tool_executors={...},
)
agent.load_skill(custom_skill)

session = agent.get_session("user-42")
print(f"Messages: {len(session.messages)}")
print(f"Last active: {session.last_activity}")

agent.clear_session("user-42")  # History cleared; session still exists

agent = Agent(llm="openai", model="gpt-4o-mini", tools=[lookup_order])
agent.set_auto_approve_all(True)

# Two independent users, one agent
response_alice = await agent.chat("Where is order #1234?", session_id="alice")
response_bob   = await agent.chat("Where is order #5678?", session_id="bob")

# Each session retains its own history
alice_session = agent.get_session("alice")
print(alice_session.messages)  # includes alice's turns only

# Clear alice's session when her conversation ends
agent.clear_session("alice")

agent.set_auto_approve_all(True)   # bypass — useful for dev/demo
agent.set_auto_approve_all(False)  # restore default (require callbacks)

agent.set_callbacks(
    on_tool_start=lambda sid, name, args: print(f"Starting {name}"),
    on_tool_end=lambda sid, name, result: print(f"Done {name}"),
    on_final_message=lambda sid, content: save_to_db(sid, content),
)

from logicore.agents.agent import Agent
import asyncio

def get_time() -> str:
    """Return the current server time in HH:MM:SS format."""
    from datetime import datetime
    return datetime.now().strftime("%H:%M:%S")

agent = Agent(llm="ollama", tools=[get_time])
agent.set_auto_approve_all(True)

response = asyncio.run(agent.chat("What time is it?"))
print(response)

def on_token(token: str):
    print(token, end="", flush=True)

agent = Agent(llm="openai", model="gpt-4o-mini")

response = asyncio.run(
    agent.chat(
        "Explain Python event loops in detail",
        stream=True,
        callbacks={"on_token": on_token},
    )
)
print("\n--- Final ---")
print(response)

async def approve_tool(session_id: str, tool_name: str, args: dict) -> bool:
    """Block destructive tools; allow everything else."""
    blocked = {"delete_file", "execute_command", "drop_table"}
    if tool_name in blocked:
        print(f"[BLOCKED] {tool_name} called with {args}")
        return False
    return True

agent = Agent(
    llm="ollama",
    tools=[read_file, write_file, delete_file],
    memory=True,
)
agent.set_callbacks(on_tool_approval=approve_tool)

response = asyncio.run(agent.chat("Clean temp files and write a summary."))
print(response)

agent = Agent(llm="openai", model="gpt-4o-mini")
agent.set_auto_approve_all(True)

# Isolated sessions per user
await agent.chat("My name is Alice.", session_id="alice")
await agent.chat("My name is Bob.",   session_id="bob")

r_alice = await agent.chat("What is my name?", session_id="alice")  # "Alice"
r_bob   = await agent.chat("What is my name?", session_id="bob")    # "Bob"

# Inspect session state
alice = agent.get_session("alice")
print(f"Alice has {len(alice.messages)} messages")

# Reset alice's conversation
agent.clear_session("alice")

agent = Agent(llm="ollama", memory=True)

# Session 1 — store a fact
await agent.chat(
    "The production database timeout is 30 seconds.",
    session_id="onboarding",
)

# Session 2 — retrieve the fact on demand via the memory tool
agent2 = Agent(llm="ollama", memory=True)
response = await agent2.chat(
    "What is the production database timeout?",
    session_id="qa-session",
)
print(response)  # Agent retrieves the fact via RAG memory tool

response = await agent.chat(
    "Analyze the sales data and write a report to report.md",
    generate_walkthrough=True,
)
# response includes the final answer + "### Walkthrough Summary" section
print(response)

# Get the step-by-step log from the last chat() call
log: list[str]   = agent.get_execution_summary()
text: str        = agent.print_execution_summary()
as_dict: dict    = agent.get_execution_summary_dict()
as_json: str     = agent.get_execution_summary_json()