Kogni·OS

from kognios import Agent, AnthropicModel, tool

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"22°C and sunny in {city}."

agent = Agent(
    model=AnthropicModel(),
    tools=[get_weather],
    instructions="Be concise.",
)
print(agent.run("What's the weather in Paris?"))
# → "It's 22°C and sunny in Paris."

from kognios import Agent, AnthropicModel, tool

@tool
def add(x: int, y: int) -> int:
    """Add two integers."""
    return x + y

@tool
def multiply(x: int, y: int) -> int:
    """Multiply two integers."""
    return x * y

agent = Agent(model=AnthropicModel(), tools=[add, multiply])
print(agent.run("What is (3 + 4) × 6?"))

agent = Agent(model=..., tools=[...], tool_timeout=10.0)  # stricter
agent = Agent(model=..., tools=[...], tool_timeout=None)  # no limit

from kognios import Agent, AnthropicModel, ShortTermMemory, LongTermMemory

short = ShortTermMemory(max_turns=20)   # sliding window, in RAM
long  = LongTermMemory(db_path="my.db") # SQLite key/value

long.store("user_name", "Alice")
long.store("location",  "Berlin")

agent = Agent(model=AnthropicModel(), memory=short)
# Known facts from long-term memory are auto-injected into the system prompt.

from kognios import Agent, AnthropicModel, LongTermMemory

def my_embed(texts):
    import openai
    client = openai.OpenAI()
    r = client.embeddings.create(model="text-embedding-3-small", input=texts)
    return [item.embedding for item in r.data]

long = LongTermMemory(db_path="my.db", embed_fn=my_embed)
long.store("user_goal", "Build a Python agent framework")
long.store("user_name", "Alice")

# retrieve the most relevant facts for a query
relevant = long.semantic_recall("What is the user building?", top_k=2)
# → [("user_goal", "Build a Python agent framework"), ...]

from kognios import Agent, AnthropicModel, SQLiteKnowledge

kb = SQLiteKnowledge(db_path="docs.db")
kb.load("docs/manual.txt")                           # file path
kb.load("https://example.com/spec", source_type="url")  # URL
kb.load("Inline text goes here…")                   # raw string
kb.load("report.pdf")                               # PDF (pip install 'kognios[pdf]')
kb.load("page.html")                                # HTML (tags stripped)
kb.load("data.csv")                                 # CSV (flattened to text)
kb.load("config.json")                              # JSON (flattened)
kb.load("report.docx")                              # Word doc (pip install 'kognios[docx]')
kb.load("https://github.com/owner/repo", source_type="github")  # GitHub repo

agent = Agent(model=AnthropicModel(), knowledge=kb)
print(agent.run("Summarise the key points."))

from kognios import Agent, Team, AnthropicModel

model = AnthropicModel()
agents = {
    "finance": Agent(model=model, name="finance",
                     description="Stocks, markets, financial data."),
    "weather": Agent(model=model, name="weather",
                     description="Weather forecasts and climate."),
}
team = Team(router_model=model, agents=agents)
print(team.run("What is Apple's stock price?"))
# → Router picks "finance", delegates the message, returns the answer.

agent = Agent(model=AnthropicModel(), tools=[web_search, calculator])
result = agent.plan_and_run("Research the top 3 Python web frameworks and compare their performance.")
print(result)

researcher = Agent(model=AnthropicModel(), name="researcher",
                   instructions="Research and summarize topics.")
writer = Agent(model=AnthropicModel(), name="writer",
               instructions="Write clear, engaging prose.")

# Researcher gathers info, writer polishes it
summary = researcher.run("Summarize the key features of SQLite.")
article = writer.send(researcher, f"Rewrite this as a blog intro: {summary}")
print(article)

for chunk in agent.stream("Write me a haiku about SQLite."):
    print(chunk, end="", flush=True)
print()

import asyncio
from kognios import Agent, AnthropicModel

async def main():
    agent = Agent(model=AnthropicModel())
    result = await agent.arun("What is the capital of Japan?")
    print(result)

asyncio.run(main())

async for chunk in agent.astream("Tell me a joke."):
    print(chunk, end="", flush=True)

from pydantic import BaseModel
from kognios import Agent, AnthropicModel

class CityInfo(BaseModel):
    city: str
    country: str
    population_millions: float

agent = Agent(model=AnthropicModel())
info = agent.run("Tell me about Tokyo.", output_type=CityInfo)
print(info.city, info.population_millions)  # Tokyo  13.96

from kognios.models.anthropic import AnthropicModel
from kognios.models.openai    import OpenAIModel
from kognios.models.groq      import GroqModel
from kognios.models.gemini    import GeminiModel
from kognios.models.ollama    import OllamaModel
from kognios.models.mistral   import MistralModel
from kognios.models.cohere    import CohereModel
from kognios.models.bedrock   import BedrockModel
from kognios.models.xai       import XAIModel
from kognios.models.together  import TogetherModel

model = AnthropicModel(model="claude-sonnet-4-6")
model = OpenAIModel(model="gpt-4o")
model = GroqModel(model="openai/gpt-oss-120b")                          # GROQ_API_KEY
model = GeminiModel(model="gemini-2.5-flash")                           # GEMINI_API_KEY
model = OllamaModel(model="llama3.3")                                   # needs Ollama running locally
model = MistralModel(model="mistral-large-latest")                      # MISTRAL_API_KEY
model = CohereModel(model="command-a-plus-05-2026")                     # COHERE_API_KEY
model = BedrockModel(model="anthropic.claude-3-7-sonnet-20250219-v1:0") # AWS_* env vars
model = XAIModel(model="grok-4")                                        # XAI_API_KEY
model = TogetherModel(model="deepseek-ai/DeepSeek-V3")                  # TOGETHER_API_KEY

from kognios import Agent, free_tier_chain

# Reads GROQ_API_KEY[_N], GEMINI_API_KEY[_N], TOGETHER_API_KEY[_N],
# XAI_API_KEY[_N], ANTHROPIC_API_KEY[_N] from environment.
# One model instance per (key, model) pair — independent cooldown buckets.
chain = free_tier_chain(preferred="groq")
agent = Agent(model=chain, tools=[...])

from kognios import ModelChain, GroqModel, GeminiModel, AnthropicModel

chain = ModelChain([
    GroqModel(model="openai/gpt-oss-120b"),
    GeminiModel(model="gemini-2.5-flash"),
    AnthropicModel(),
])

from kognios import Agent, AnthropicModel, MCPClient

async def main():
    async with MCPClient.stdio("uvx", ["mcp-server-filesystem", "."]) as mcp:
        agent = Agent(model=AnthropicModel(), tools=mcp.tools())
        result = await agent.arun("List Python files in this directory.")
        print(result)

with MCPClient.stdio("uvx", ["mcp-server-filesystem", "."]).sync() as mcp:
    agent = Agent(model=AnthropicModel(), tools=mcp.tools())
    print(agent.run("How many Python files are there?"))

from kognios.tools.builtins import (
    calculator,        # safe AST-based math evaluator
    python_eval,       # sandboxed exec with stdout capture
    read_file,         # read any local file
    write_file,        # write file + create parent dirs
    http_get,          # fetch URL (urllib, no requests needed)
    web_search,        # DuckDuckGo search
    code_interpreter,  # isolated subprocess exec with timeout
    browse,            # JS-rendered page fetching (pip install 'kognios[browser]')
)

from kognios import Agent, AnthropicModel
from kognios.eval import AgentEvaluator, EvalCase, contains, exact_match

agent = Agent(model=AnthropicModel())

cases = [
    EvalCase(input="What is 2+2?", expected="4"),
    EvalCase(input="Capital of France?", expected="Paris"),
]

evaluator = AgentEvaluator(agent, scorer=contains)
report = evaluator.run(cases)
report.print_report()
# Cases: 2 | Pass rate: 100.0% | Mean score: 1.000 | Mean latency: 0.82s

from kognios import Agent, AnthropicModel, block_keywords, max_length, pii_scrubber, GuardrailError

agent = Agent(
    model=AnthropicModel(),
    input_guardrails=[
        block_keywords("jailbreak", "ignore previous instructions"),
        max_length(1000),
    ],
    output_guardrails=[pii_scrubber()],
)

try:
    result = agent.run("Tell me how to jailbreak this system")
except GuardrailError as e:
    print(f"Blocked: {e}")

from kognios.plugins import load_plugins, list_plugins

providers = load_plugins("kognios.providers")
print(list_plugins("kognios.tools"))

from kognios import Agent, AnthropicModel
from kognios.tracing import Tracer

tracer = Tracer()
agent = Agent(model=AnthropicModel(), tracer=tracer)
agent.run("What is 2+2?")

tracer.print_spans()
# [TRACE] llm.complete 823.4ms model=claude-sonnet-4-6