# Directory Structure
```
├── .gitignore
├── .python-version
├── AGENTS.md
├── CONTRIBUTING.md
├── examples
│ └── basic_mission.py
├── pyproject.toml
├── README.md
├── src
│ └── stk_mcp
│ ├── __init__.py
│ ├── app.py
│ ├── cli.py
│ ├── stk_logic
│ │ ├── analysis.py
│ │ ├── config.py
│ │ ├── core.py
│ │ ├── decorators.py
│ │ ├── location.py
│ │ ├── logging_config.py
│ │ ├── objects.py
│ │ ├── satellite.py
│ │ ├── scenario.py
│ │ └── utils.py
│ └── tools
│ ├── __init__.py
│ ├── analysis.py
│ ├── health.py
│ ├── location.py
│ ├── objects.py
│ ├── satellite.py
│ └── scenario.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
1 | 3.12
2 |
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | # Python-generated files
2 | __pycache__/
3 | *.py[oc]
4 | build/
5 | dist/
6 | wheels/
7 | *.egg-info
8 |
9 | # Virtual environments
10 | .venv
11 | *.whl
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # STK-MCP
2 |
3 | [](https://www.python.org/downloads/) [](https://pypi.org/project/mcp/)
4 |
5 | STK-MCP is an MCP (Model Context Protocol) server designed to enable Large Language Models (LLMs) or other MCP clients to interact with [Ansys/AGI STK](https://www.ansys.com/products/missions/ansys-stk) (Systems Tool Kit) - the leading Digital Mission Engineering software.
6 |
7 |
8 | This project allows controlling STK via an MCP server, supporting both STK Desktop (Windows only) and STK Engine (Windows & Linux). It utilizes `FastMCP` from the official [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk).
9 |
10 | ## Overview
11 |
12 | The primary goal of this project is to bridge the gap between programmatic interaction and the powerful simulation capabilities of STK. By exposing STK functionalities through a robust CLI and an MCP server, users can command STK simulations using simple commands or LLM-driven applications.
13 |
14 | The MCP application, defined in `src/stk_mcp/app.py`, exposes STK operations as MCP tools, which are dynamically managed by a CLI entry point in `src/stk_mcp/cli.py`.
15 |
16 | ## Features
17 |
18 | * CLI entry point powered by `Typer`.
19 | * Dual mode operation: STK Engine (Windows/Linux) and STK Desktop (Windows).
20 | * OS-aware: Desktop mode auto-disabled on non-Windows platforms.
21 | * Managed lifecycle: STK instance is started/stopped with the MCP server.
22 | * Tool discovery: `list-tools` command enumerates available MCP tools.
23 | * Modular architecture: CLI (`cli.py`), MCP (`app.py`), STK logic (`stk_logic/`), and MCP tools (`tools/`).
24 |
25 | ## Prerequisites
26 |
27 | * **Operating System:** Windows or Linux. STK Desktop mode requires Windows.
28 | * **Python:** Version 3.12 or higher.
29 | * **Ansys/AGI STK:** Version 12.x Desktop or Engine installed.
30 | * **STK Python API:** The `agi.stk12` Python wheel corresponding to your STK installation must be available. Typically found under `CodeSamples\Automation\Python` in your STK install.
31 |
32 | ## Installation
33 |
34 | 1. Clone the repository
35 | ```bash
36 | git clone <repository-url>
37 | cd stk-mcp
38 | ```
39 | 2. Create and activate a virtual environment
40 | ```bash
41 | # Create the virtual environment
42 | uv venv
43 |
44 | # Activate it
45 | # On Windows (in PowerShell/CMD):
46 | # .venv\Scripts\activate
47 | # On Linux (in bash/zsh):
48 | source .venv/bin/activate
49 | ```
50 | 3. Add dependencies with uv (preferred)
51 | - Add the STK Python wheel from your STK installation (local file):
52 | ```bash
53 | uv add ./agi.stk12-12.10.0-py3-none-any.whl
54 | # or: uv add path/to/your/STK/CodeSamples/Automation/Python/agi.stk12-*.whl
55 |
56 | # Windows only: COM bridge for Desktop automation
57 | uv add "pywin32; platform_system == 'Windows'"
58 | ```
59 | 4. Sync the environment (installs deps from `pyproject.toml`)
60 | ```bash
61 | uv sync
62 | ```
63 |
64 | ## Usage
65 |
66 | This project is a command-line application. Ensure your virtual environment is activated before running commands.
67 |
68 | ### Listing Available Tools
69 |
70 | ```bash
71 | uv run -m stk_mcp.cli list-tools
72 | ```
73 | Prints a table of tool names and their descriptions.
74 |
75 | ### Running the MCP Server
76 |
77 | Use the `run` command to start the MCP server. The server will automatically start and manage an STK instance.
78 |
79 | Run with `uv run` so you don’t need to install the package into site-packages.
80 |
81 | **1) STK Engine (recommended for automation, Windows/Linux):**
82 | ```bash
83 | uv run -m stk_mcp.cli run --mode engine
84 | ```
85 |
86 | **2) STK Desktop (Windows only, shows GUI):**
87 | Ensure STK Desktop is closed; the server will launch and manage its own instance.
88 | ```bash
89 | uv run -m stk_mcp.cli run --mode desktop
90 | ```
91 |
92 | The server will start, initialize STK, and listen for MCP connections on `http://127.0.0.1:8765` by default.
93 |
94 | **3. Command Options:**
95 | You can see all options with the `--help` flag:
96 | ```bash
97 | stk-mcp run --help
98 | ```
99 |
100 | ### Interacting with the Server
101 | Once the server is running, you can connect to it using any MCP client, such as the MCP Inspector.
102 |
103 | 1. Open the MCP Inspector URL provided in the console (e.g., `http://127.0.0.1:8765`).
104 | 2. Find the "STK Control" server in the list.
105 | 3. Use the "Tools" section to execute `setup_scenario`, `create_location`, and `create_satellite`.
106 |
107 | ### Stopping the Server
108 | Press `Ctrl+C` in the terminal where the server is running. The lifecycle manager will automatically close the STK Engine or Desktop instance.
109 |
110 | ## MCP Tools and Resources
111 |
112 | The server exposes the following MCP tools/resources.
113 |
114 | | Name | Kind | Description | Desktop (Windows) | Engine (Windows) | Engine (Linux) |
115 | |------------------|----------|-----------------------------------------------------------------------------------------------|-------------------|------------------|----------------|
116 | | `setup_scenario` | Tool | Create/configure an STK Scenario; sets time period and rewinds animation. | Yes | Yes | Yes |
117 | | `create_location`| Tool | Create/update a `Facility` (default) or `Place` at latitude/longitude/altitude (km). | Yes | Yes | Yes |
118 | | `create_satellite`| Tool | Create/configure a satellite from apogee/perigee (km), RAAN, and inclination; TwoBody prop. | Yes | Yes | No |
119 |
120 | Notes:
121 | - `create_satellite` on Linux Engine is not yet supported because it relies on COM-specific casts; a Connect-based fallback is planned.
122 |
123 | Resources:
124 |
125 | | Name | Kind | Description | Desktop (Windows) | Engine (Windows) | Engine (Linux) |
126 | |------|------|-------------|-------------------|------------------|----------------|
127 | | `resource://stk/objects` | Resource | List all objects in the active scenario. Returns JSON records: `{name, type}`. | Yes | Yes | Yes |
128 | | `resource://stk/objects/{type}` | Resource | List objects filtered by `type` (e.g., `satellite`, `facility`, `place`, `sensor`). Returns JSON records. | Yes | Yes | Yes |
129 | | `resource://stk/health` | Resource | Report basic state: mode, scenario name, and object counts. | Yes | Yes | Yes |
130 | | `resource://stk/analysis/access/{object1}/{object2}` | Resource | Compute access intervals between two objects. Provide paths like `Satellite/SatA` and `Facility/FacB` (with or without leading `*/`). | Yes | Yes | Yes |
131 | | `resource://stk/reports/lla/{satellite}` | Resource | Return satellite LLA ephemeris over the scenario start/stop interval. Provide path like `Satellite/SatA` (with or without leading `*/`). | Yes | Yes | Yes |
132 |
133 | Examples:
134 |
135 | - Read all objects: `resource://stk/objects`
136 | - Read only satellites: `resource://stk/objects/satellite`
137 | - Read ground locations: `resource://stk/objects/location` (alias for facilities and places)
138 |
139 | Access and LLA examples:
140 |
141 | - Compute access: `resource://stk/analysis/access/Satellite/ISS/Facility/Boulder`
142 | - Get ISS LLA (60 s): `resource://stk/reports/lla/Satellite/ISS` (optional `step_sec` argument)
143 |
144 | ## Configuration & Logging
145 |
146 | Configuration is centralized in `src/stk_mcp/stk_logic/config.py` using `pydantic-settings`.
147 | Defaults can be overridden with environment variables (prefix `STK_MCP_`).
148 |
149 | - `STK_MCP_DEFAULT_HOST` (default `127.0.0.1`)
150 | - `STK_MCP_DEFAULT_PORT` (default `8765`)
151 | - `STK_MCP_LOG_LEVEL` (default `INFO`)
152 | - `STK_MCP_DEFAULT_SCENARIO_NAME` (default `MCP_STK_Scenario`)
153 | - `STK_MCP_DEFAULT_START_TIME` (default `20 Jan 2020 17:00:00.000`)
154 | - `STK_MCP_DEFAULT_DURATION_HOURS` (default `48.0`)
155 |
156 | Logging is standardized via `src/stk_mcp/stk_logic/logging_config.py`. The CLI uses
157 | this configuration, producing structured logs with timestamps, levels, and context.
158 |
159 | ## Implementation Notes
160 |
161 | - STK access is serialized with a global lock to avoid concurrency issues.
162 | - Common STK-availability checks are handled via decorators in
163 | `src/stk_mcp/stk_logic/decorators.py` (`@require_stk_tool` and `@require_stk_resource`).
164 | - STK Connect commands that may be transiently flaky are executed with retry logic
165 | (`tenacity`) in `src/stk_mcp/stk_logic/utils.py` (`safe_stk_command`).
166 | - Long-running internal operations are timed with `@timed_operation` for diagnostics.
167 |
168 | ## Dependencies
169 |
170 | Managed with `uv`:
171 |
172 | * `agi.stk12` (local wheel from your STK install)
173 | * `mcp[cli]>=1.6.0`
174 | * `uvicorn>=0.30` (explicit for CLI server)
175 | * `rich>=13.7` (CLI table output)
176 | * `typer>=0.15.2`
177 | * `pydantic>=2.11.7`
178 | * `pywin32` (Windows only)
179 |
180 | Notes:
181 | - On macOS (Darwin), STK Engine/Desktop are not supported. The server will start but STK-dependent tools/resources are unavailable.
182 | - The server serializes STK access via a global lock to avoid concurrency issues with COM/Engine calls.
183 |
184 | ## Contributing
185 |
186 | Contributions are welcome! Please see the [CONTRIBUTING.md](CONTRIBUTING.md) file for guidelines.
187 |
```
--------------------------------------------------------------------------------
/AGENTS.md:
--------------------------------------------------------------------------------
```markdown
1 | You are a senior Python software engineer and a domain expert in Orbital Mechanics and Ansys/AGI STK (Satellite/Systems Tool Kit).
2 | You are tasked to work on this project called STK-MCP that allows LLMs to interact with Ansys/AGI STK (Satellite/Systems Tool Kit) using the Model Context Protocol (MCP).
3 |
4 | # Guidelines
5 | - always use `uv` to manage the project and dependencies.
6 | - always use `uv add <package_name>` to add packages/dependencies instead of `pip install`.
7 | - to remove a package/dependency, always use `uv remove <package_name>`.
8 | - always use `uv sync` to create/update the virtual environment with the dependencies in `pyproject.toml`.
9 | - always use `uv run` to run the project.
10 | - always aim to make the mcp tools/resources applicalbe with both STK Desktop and STK Engine.
11 | - exceptions should be made for tools/resources that are only applicable to one of the modes.
12 | - when adding a new mcp tool/resource, always:
13 | - add it to the `tools` directory
14 | - add any STK SDK logic code to the `stk_logic` directory
15 | - add the necessary documentation in the `README.md` file (update tools and/or resources table).
```
--------------------------------------------------------------------------------
/CONTRIBUTING.md:
--------------------------------------------------------------------------------
```markdown
1 | # Contributing to STK-MCP
2 |
3 | First off, thank you for considering contributing to STK-MCP! Your help is appreciated.
4 |
5 | This document provides guidelines for contributing to the project.
6 |
7 | ## How Can I Contribute?
8 |
9 | There are several ways you can contribute:
10 |
11 | * **Reporting Bugs:** If you find a bug, please open an issue in the GitHub repository. Describe the bug clearly, including steps to reproduce it, expected behavior, and actual behavior. Include STK version, Python version, and OS information if relevant.
12 | * **Suggesting Enhancements:** Have an idea for a new feature or an improvement to an existing one? Open an issue to discuss it. Provide as much detail as possible about your suggestion.
13 | * **Writing Code:** If you'd like to contribute code, please follow the steps below.
14 | * **Improving Documentation:** If you find errors or areas for improvement in the README or other documentation, feel free to suggest changes.
15 |
16 | ## Development Process
17 |
18 | 1. **Fork the repository:** Create your own fork of the `stk-mcp` repository on GitHub.
19 | 2. **Clone your fork:** `git clone https://github.com/YOUR_USERNAME/stk-mcp.git`
20 | 3. **Create a branch:** `git checkout -b feature/your-feature-name` or `git checkout -b fix/your-bug-fix-name`
21 | 4. **Make your changes:** Implement your feature or bug fix.
22 | * Follow the existing code style.
23 | * Add comments for complex logic.
24 | * Ensure your changes work with the required versions of Python and STK.
25 | * Update documentation if necessary.
26 | 5. **Test your changes:** Ensure your changes don't break existing functionality. (Add details here if specific tests are set up later).
27 | 6. **Commit your changes:** Use clear and descriptive commit messages. `git commit -m "feat: Add feature X"` or `git commit -m "fix: Resolve issue Y"`
28 | 7. **Push to your fork:** `git push origin feature/your-feature-name`
29 | 8. **Open a Pull Request:** Go to the original `stk-mcp` repository on GitHub and open a pull request from your branch to the `main` branch (or the appropriate target branch).
30 | * Provide a clear title and description for your pull request, explaining the changes and why they are needed.
31 | * Link to any relevant issues.
32 |
33 | ## Pull Request Guidelines
34 |
35 | * Ensure your code adheres to the project's style guidelines (details can be added here, e.g., linting tools like Black, Flake8).
36 | * Keep pull requests focused on a single feature or bug fix.
37 | * Provide a clear description of the changes.
38 | * Be responsive to feedback and questions during the review process.
39 |
40 | Thank you for contributing!
```
--------------------------------------------------------------------------------
/src/stk_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """STK-MCP package.
2 |
3 | This package contains the MCP server (`app`), CLI (`cli`), tools, and
4 | STK-specific logic under `stk_logic`.
5 | """
6 |
7 | # Optionally expose the server at package level if desired:
8 | # from .app import mcp_server
9 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/app.py:
--------------------------------------------------------------------------------
```python
1 | from mcp.server.fastmcp import FastMCP
2 | from .stk_logic.core import StkState
3 |
4 | # Define the central MCP server instance.
5 | # The lifespan will be attached dynamically by the CLI based on user selection.
6 | mcp_server = FastMCP[StkState]("STK Control")
7 |
8 | # Import the tools module.
9 | # This triggers the execution of the @mcp_server.tool() decorators.
10 | from . import tools
11 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """Register MCP tools and resources.
2 |
3 | Importing modules ensures the @mcp_server.tool() and .resource() decorators
4 | run and register endpoints with the server.
5 | """
6 |
7 | from . import scenario # noqa: F401
8 | from . import satellite # noqa: F401
9 | from . import location # noqa: F401
10 | from . import objects # noqa: F401
11 | from . import health # noqa: F401
12 | from . import analysis # noqa: F401
13 |
14 | # You can optionally define an __all__ if needed, but importing is usually sufficient
15 | # for the decorators to register.
16 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/logging_config.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | import logging
4 | import sys
5 |
6 |
7 | def configure_logging(level: str = "INFO") -> None:
8 | """Configure structured logging for STK-MCP.
9 |
10 | Example format:
11 | 2025-01-01 12:00:00 | INFO | stk_mcp.module:function:42 - Message
12 | """
13 | fmt = (
14 | "%(asctime)s | %(levelname)-8s | "
15 | "%(name)s:%(funcName)s:%(lineno)d - %(message)s"
16 | )
17 | datefmt = "%Y-%m-%d %H:%M:%S"
18 |
19 | handler = logging.StreamHandler(sys.stdout)
20 | handler.setFormatter(logging.Formatter(fmt=fmt, datefmt=datefmt))
21 |
22 | root = logging.getLogger()
23 | for h in list(root.handlers):
24 | root.removeHandler(h)
25 | root.addHandler(handler)
26 |
27 | lvl = getattr(logging, str(level).upper(), logging.INFO)
28 | root.setLevel(lvl)
29 |
30 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "stk-mcp"
3 | version = "0.2.0"
4 | description = "STK-MCP, an MCP server allowing LLMs to interact with Ansys/AGI STK - Digital Mission Engineering Software"
5 | readme = "README.md"
6 | requires-python = ">=3.12"
7 | dependencies = [
8 | "mcp[cli]>=1.6.0",
9 | "pydantic>=2.11.7",
10 | "typer>=0.15.2",
11 | "uvicorn>=0.30",
12 | "rich>=13.7",
13 | "pywin32>=306; platform_system == 'Windows'",
14 | "agi-stk12",
15 | "pydantic-settings>=2.10.1",
16 | "tenacity>=9.1.2",
17 | ]
18 |
19 | [project.scripts]
20 | stk-mcp = "stk_mcp.cli:app" # New CLI entry point
21 |
22 | # Optional: Add pywin32 if needed for clarity, though pip usually handles it on Windows
23 | # dependencies = [
24 | # "mcp[cli]>=1.6.0",
25 | # "pywin32>=306; sys_platform == 'win32'" # Example conditional dependency
26 | # ]
27 |
28 | [build-system]
29 | requires = ["uv-build>=0.4.0"]
30 | build-backend = "uv_build"
31 |
32 | [tool.uv.sources]
33 | agi-stk12 = { path = "agi.stk12-12.10.0-py3-none-any.whl" }
34 |
```
--------------------------------------------------------------------------------
/examples/basic_mission.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Example: Creating a basic Earth observation mission.
3 |
4 | This demonstrates:
5 | 1. Setting up a scenario
6 | 2. Creating a ground facility
7 | 3. Creating a simple satellite (Windows/Desktop/Engine on Windows)
8 |
9 | Run the server first, then invoke these tools from your MCP client.
10 | """
11 |
12 | # This file is illustrative. Use an MCP client (e.g., MCP Inspector)
13 | # to call the following tools in order:
14 | #
15 | # 1) setup_scenario(scenario_name="Demo", start_time="20 Jan 2020 17:00:00.000", duration_hours=12)
16 | # 2) create_location(name="Boulder", latitude_deg=40.015, longitude_deg=-105.27, altitude_km=1.656, kind="facility")
17 | # 3) create_satellite(name="DemoSat", apogee_alt_km=420, perigee_alt_km=410, raan_deg=51.6, inclination_deg=51.6)
18 | # 4) resource: resource://stk/analysis/access/Satellite/DemoSat/Facility/Boulder
19 | # 5) resource: resource://stk/reports/lla/Satellite/DemoSat
20 |
21 | print(
22 | "Open the MCP client and try the listed tools/resources to run this example."
23 | )
24 |
25 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/config.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | """
4 | Central configuration for STK-MCP.
5 |
6 | Values can be overridden via environment variables prefixed with `STK_MCP_`.
7 | For example: `STK_MCP_DEFAULT_PORT=9000`.
8 | """
9 |
10 | from pydantic_settings import BaseSettings, SettingsConfigDict
11 |
12 |
13 | class StkConfig(BaseSettings):
14 | """STK-MCP configuration.
15 |
16 | Environment variables use the `STK_MCP_` prefix, e.g., `STK_MCP_DEFAULT_PORT`.
17 | """
18 |
19 | # Physical constants / domain
20 | earth_radius_km: float = 6378.137
21 |
22 | # Scenario defaults
23 | default_scenario_name: str = "MCP_STK_Scenario"
24 | default_start_time: str = "20 Jan 2020 17:00:00.000"
25 | default_duration_hours: float = 48.0
26 |
27 | # Server defaults
28 | default_host: str = "127.0.0.1"
29 | default_port: int = 8765
30 |
31 | # Logging
32 | log_level: str = "INFO"
33 |
34 | model_config = SettingsConfigDict(
35 | env_prefix="STK_MCP_",
36 | extra="ignore",
37 | )
38 |
39 |
40 | def get_config() -> StkConfig:
41 | """Return a cached settings instance."""
42 | # pydantic-settings already caches by default, but we can wrap for clarity
43 | return StkConfig() # type: ignore[call-arg]
44 |
45 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/health.py:
--------------------------------------------------------------------------------
```python
1 | import logging
2 | from collections import Counter
3 | from mcp.server.fastmcp import Context
4 | from mcp.server.fastmcp.exceptions import ResourceError
5 |
6 | from ..app import mcp_server
7 | from ..stk_logic.core import StkState, STK_LOCK
8 | from ..stk_logic.decorators import require_stk_resource
9 | from ..stk_logic.objects import list_objects_internal
10 |
11 | logger = logging.getLogger(__name__)
12 |
13 |
14 | @mcp_server.resource(
15 | "resource://stk/health",
16 | name="STK Health",
17 | title="STK Server Health",
18 | description=(
19 | "Report basic STK state: mode, current scenario, and object counts."
20 | ),
21 | mime_type="application/json",
22 | )
23 | @require_stk_resource
24 | def health(ctx: Context):
25 | lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
26 |
27 | if not lifespan_ctx:
28 | raise ResourceError("No lifespan context set.")
29 |
30 | mode = lifespan_ctx.mode.value if lifespan_ctx.mode else None
31 | scenario_name = None
32 | try:
33 | if lifespan_ctx.stk_root and lifespan_ctx.stk_root.CurrentScenario:
34 | scenario_name = lifespan_ctx.stk_root.CurrentScenario.InstanceName
35 | except Exception:
36 | scenario_name = None
37 |
38 | objects = []
39 | try:
40 | if lifespan_ctx.stk_root:
41 | with STK_LOCK:
42 | objects = list_objects_internal(lifespan_ctx.stk_root)
43 | except Exception:
44 | objects = []
45 |
46 | counts = Counter([o.get("type", "") for o in objects if o.get("type")])
47 |
48 | return {
49 | "mode": mode,
50 | "scenario": scenario_name,
51 | "counts": dict(counts),
52 | }
53 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/utils.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | import logging
4 | import time
5 | from functools import wraps
6 | from typing import Any, Callable, TypeVar, ParamSpec
7 |
8 | from tenacity import retry, stop_after_attempt, wait_exponential
9 |
10 | logger = logging.getLogger(__name__)
11 |
12 | P = ParamSpec("P")
13 | T = TypeVar("T")
14 |
15 |
16 | def timed_operation(func: Callable[P, T]) -> Callable[P, T]:
17 | """Decorator to log operation duration for diagnostics."""
18 |
19 | @wraps(func)
20 | def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
21 | start = time.perf_counter()
22 | try:
23 | result = func(*args, **kwargs)
24 | duration = time.perf_counter() - start
25 | logger.info("%s completed in %.3fs", func.__name__, duration)
26 | return result
27 | except Exception as e: # pragma: no cover - diagnostic path
28 | duration = time.perf_counter() - start
29 | logger.error("%s failed after %.3fs: %s", func.__name__, duration, e)
30 | raise
31 |
32 | return wrapper
33 |
34 |
35 | @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), reraise=True)
36 | def safe_stk_command(stk_root: Any, command: str):
37 | """Execute an STK Connect command with basic retry logic."""
38 | return stk_root.ExecuteCommand(command)
39 |
40 |
41 | def safe_exec_lines(stk_root: Any, command: str) -> list[str]:
42 | """Execute a Connect command and return result lines; returns [] on failure."""
43 | try:
44 | res = safe_stk_command(stk_root, command)
45 | return [res.Item(i) for i in range(res.Count)]
46 | except Exception: # pragma: no cover - depends on STK runtime
47 | logger.debug("Connect command failed: %s", command)
48 | return []
49 |
50 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/decorators.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | import logging
4 | from functools import wraps
5 | from typing import Callable, Any, TypeVar, ParamSpec
6 |
7 | from mcp.server.fastmcp import Context
8 | from mcp.server.fastmcp.exceptions import ResourceError
9 |
10 | from .core import stk_available
11 |
12 | logger = logging.getLogger(__name__)
13 |
14 | P = ParamSpec("P")
15 | T = TypeVar("T")
16 |
17 |
18 | def require_stk_tool(func: Callable[P, T]) -> Callable[P, T]:
19 | """Ensure STK is available and initialized for MCP tools.
20 |
21 | Returns a user-friendly error string if unavailable.
22 | Expects first parameter to be `ctx: Context`.
23 | """
24 |
25 | @wraps(func)
26 | def wrapper(ctx: Context, *args: P.args, **kwargs: P.kwargs) -> T: # type: ignore[override]
27 | lifespan_ctx = ctx.request_context.lifespan_context
28 |
29 | if not stk_available:
30 | return "Error: STK is not available on this system." # type: ignore[return-value]
31 | if not lifespan_ctx or not lifespan_ctx.stk_root:
32 | return "Error: STK Root not available. Initialize via server lifespan." # type: ignore[return-value]
33 |
34 | return func(ctx, *args, **kwargs)
35 |
36 | return wrapper
37 |
38 |
39 | def require_stk_resource(func: Callable[P, T]) -> Callable[P, T]:
40 | """Ensure STK is available and initialized for MCP resources.
41 |
42 | Raises ResourceError if unavailable. Expects first parameter `ctx: Context`.
43 | """
44 |
45 | @wraps(func)
46 | def wrapper(ctx: Context, *args: P.args, **kwargs: P.kwargs) -> T: # type: ignore[override]
47 | lifespan_ctx = ctx.request_context.lifespan_context
48 | if not stk_available:
49 | raise ResourceError("STK is not available on this system.")
50 | if not lifespan_ctx or not lifespan_ctx.stk_root:
51 | raise ResourceError("STK Root not available. Initialize via server lifespan.")
52 | return func(ctx, *args, **kwargs)
53 |
54 | return wrapper
55 |
56 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/analysis.py:
--------------------------------------------------------------------------------
```python
1 | import logging
2 | from mcp.server.fastmcp import Context
3 | from mcp.server.fastmcp.exceptions import ResourceError
4 |
5 | from ..app import mcp_server
6 | from ..stk_logic.core import StkState, STK_LOCK
7 | from ..stk_logic.decorators import require_stk_resource
8 | from ..stk_logic.analysis import (
9 | compute_access_intervals_internal,
10 | get_lla_ephemeris_internal,
11 | )
12 |
13 | logger = logging.getLogger(__name__)
14 |
15 |
16 | @mcp_server.resource(
17 | "resource://stk/analysis/access/{object1}/{object2}",
18 | name="STK Access",
19 | title="Compute Access Intervals",
20 | description=(
21 | "Compute access intervals between two objects."
22 | " Provide paths like 'Satellite/SatA' and 'Facility/FacB'"
23 | " (with or without leading '*/')."
24 | ),
25 | mime_type="application/json",
26 | )
27 | @require_stk_resource
28 | def compute_access(ctx: Context, object1: str, object2: str):
29 | lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
30 | if not lifespan_ctx or not lifespan_ctx.stk_root:
31 | raise ResourceError("STK Root unavailable.")
32 |
33 | with STK_LOCK:
34 | return compute_access_intervals_internal(lifespan_ctx.stk_root, object1, object2)
35 |
36 |
37 | @mcp_server.resource(
38 | "resource://stk/reports/lla/{satellite}",
39 | name="STK LLA Ephemeris",
40 | title="Satellite LLA Ephemeris",
41 | description=(
42 | "Return satellite LLA ephemeris over the scenario interval."
43 | " Provide path like 'Satellite/SatA' (with or without leading '*/')."
44 | ),
45 | mime_type="application/json",
46 | )
47 | @require_stk_resource
48 | def get_satellite_lla(ctx: Context, satellite: str):
49 | lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
50 | if not lifespan_ctx or not lifespan_ctx.stk_root:
51 | raise ResourceError("STK Root unavailable.")
52 |
53 | with STK_LOCK:
54 | return get_lla_ephemeris_internal(lifespan_ctx.stk_root, satellite, 60.0)
55 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/objects.py:
--------------------------------------------------------------------------------
```python
1 | import logging
2 | from mcp.server.fastmcp import Context
3 | from mcp.server.fastmcp.exceptions import ResourceError
4 |
5 | from ..app import mcp_server
6 | from ..stk_logic.core import StkState, STK_LOCK
7 | from ..stk_logic.decorators import require_stk_resource
8 | from ..stk_logic.objects import list_objects_internal
9 |
10 | logger = logging.getLogger(__name__)
11 |
12 |
13 | @mcp_server.resource(
14 | "resource://stk/objects",
15 | name="STK Scenario Objects",
16 | title="List STK Objects",
17 | description=(
18 | "List all objects in the active STK scenario with their name and type. "
19 | "Returns JSON: [{name, type}, ...]."
20 | ),
21 | mime_type="application/json",
22 | )
23 | @require_stk_resource
24 | def list_objects(ctx: Context):
25 | """
26 | MCP Resource: List all scenario objects as JSON records.
27 | """
28 | lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
29 |
30 | try:
31 | with STK_LOCK:
32 | return list_objects_internal(lifespan_ctx.stk_root)
33 | except Exception as e:
34 | raise ResourceError(str(e))
35 |
36 |
37 | @mcp_server.resource(
38 | "resource://stk/objects/{object_type}",
39 | name="STK Scenario Objects (Filtered)",
40 | title="List STK Objects by Type",
41 | description=(
42 | "List scenario objects filtered by type (e.g., satellite, facility, place, sensor). "
43 | "Returns JSON: [{name, type}, ...]."
44 | ),
45 | mime_type="application/json",
46 | )
47 | @require_stk_resource
48 | def list_objects_by_type(ctx: Context, object_type: str):
49 | """
50 | MCP Resource: List scenario objects filtered by the provided type.
51 | """
52 | lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
53 |
54 | try:
55 | with STK_LOCK:
56 | objects = list_objects_internal(lifespan_ctx.stk_root, filter_type=object_type)
57 | # If the filter was unrecognized, return empty with a hint instead of throwing
58 | if not objects:
59 | # We still return JSON for consistency
60 | return []
61 | return objects
62 | except Exception as e:
63 | raise ResourceError(str(e))
64 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/scenario.py:
--------------------------------------------------------------------------------
```python
1 | import logging
2 | from mcp.server.fastmcp import Context
3 |
4 | # Use relative imports within the package
5 | from ..app import mcp_server # Import the server instance created in server.py
6 | from ..stk_logic.core import StkState, STK_LOCK
7 | from ..stk_logic.decorators import require_stk_tool
8 | from ..stk_logic.config import get_config
9 | from ..stk_logic.scenario import setup_scenario_internal
10 |
11 | logger = logging.getLogger(__name__)
12 |
13 | @mcp_server.tool() # Decorate with the server instance
14 | @require_stk_tool
15 | def setup_scenario(
16 | ctx: Context,
17 | scenario_name: str | None = None,
18 | start_time: str | None = None, # Default UTCG start
19 | duration_hours: float | None = None # Default duration
20 | ) -> str:
21 | """
22 | MCP Tool: Creates/Configures an STK Scenario. Closes any existing scenario first.
23 |
24 | Args:
25 | ctx: The MCP context (provides access to stk_root via lifespan).
26 | scenario_name: Name for the new scenario.
27 | start_time: Scenario start time in STK UTCG format.
28 | duration_hours: Scenario duration in hours.
29 |
30 | Returns:
31 | A string indicating success or failure.
32 | """
33 | logger.info("MCP Tool: setup_scenario '%s'", scenario_name)
34 | lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
35 |
36 | # Basic input validation
37 | cfg = get_config()
38 | if scenario_name is None:
39 | scenario_name = cfg.default_scenario_name
40 | if start_time is None:
41 | start_time = cfg.default_start_time
42 | if duration_hours is None:
43 | duration_hours = cfg.default_duration_hours
44 |
45 | if not scenario_name or not isinstance(scenario_name, str):
46 | return "Error: scenario_name must be a non-empty string."
47 | if duration_hours <= 0:
48 | return "Error: duration_hours must be positive."
49 |
50 | # Call the internal logic function
51 | with STK_LOCK:
52 | success, message, _ = setup_scenario_internal(
53 | stk_root=lifespan_ctx.stk_root,
54 | scenario_name=scenario_name,
55 | start_time=start_time,
56 | duration_hours=duration_hours,
57 | )
58 |
59 | return message # Return the status message from the internal function
60 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/scenario.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | import logging
4 | from .core import stk_available, IAgStkObjectRoot, IAgScenario
5 | from .utils import timed_operation, safe_stk_command
6 |
7 | logger = logging.getLogger(__name__)
8 |
9 | @timed_operation
10 | def setup_scenario_internal(
11 | stk_root: IAgStkObjectRoot,
12 | scenario_name: str,
13 | start_time: str,
14 | duration_hours: float
15 | ) -> tuple[bool, str, IAgScenario | None]:
16 | """
17 | Internal logic to create/configure an STK Scenario.
18 |
19 | Returns:
20 | tuple: (success_flag, status_message, scenario_object_or_None)
21 | """
22 | if not stk_available or not stk_root:
23 | return False, "STK Root object not available.", None
24 |
25 | try:
26 | # Close existing scenario if open
27 | if stk_root.Children.Count > 0:
28 | current_scen_name = stk_root.CurrentScenario.InstanceName
29 | logger.info(" Closing existing scenario: %s", current_scen_name)
30 | stk_root.CloseScenario()
31 |
32 | # Create new scenario
33 | logger.info(" Creating new scenario: %s", scenario_name)
34 | stk_root.NewScenario(scenario_name)
35 | scenario = stk_root.CurrentScenario
36 |
37 | if scenario is None:
38 | raise Exception("Failed to create or get the new scenario object.")
39 |
40 | # Set time period
41 | duration_str = f"+{duration_hours} hours"
42 | logger.info(" Setting scenario time: Start='%s', Duration='%s'", start_time, duration_str)
43 | scenario.SetTimePeriod(start_time, duration_str)
44 |
45 | # Reset animation time
46 | stk_root.Rewind()
47 |
48 | # Optional: Maximize windows
49 | try:
50 | logger.info(" Maximizing STK windows...")
51 | safe_stk_command(stk_root, 'Application / Raise')
52 | safe_stk_command(stk_root, 'Application / Maximize')
53 | # Consider checking for 3D window existence if needed
54 | # stk_root.ExecuteCommand('Window3D * Maximize')
55 | except Exception as cmd_e:
56 | logger.warning(" Could not execute maximize commands: %s", cmd_e)
57 |
58 | return True, f"Successfully created and configured scenario: '{scenario_name}'", scenario
59 |
60 | except Exception as e:
61 | error_msg = f"Error setting up scenario '{scenario_name}': {e}"
62 | logger.error(" %s", error_msg)
63 | # import traceback
64 | # traceback.print_exc()
65 | return False, error_msg, None
66 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/location.py:
--------------------------------------------------------------------------------
```python
1 | import logging
2 | from mcp.server.fastmcp import Context
3 |
4 | from ..app import mcp_server
5 | from ..stk_logic.core import StkState, STK_LOCK
6 | from ..stk_logic.decorators import require_stk_tool
7 | from ..stk_logic.location import create_location_internal
8 |
9 | logger = logging.getLogger(__name__)
10 |
11 |
12 | @mcp_server.tool()
13 | @require_stk_tool
14 | def create_location(
15 | ctx: Context,
16 | name: str,
17 | latitude_deg: float,
18 | longitude_deg: float,
19 | altitude_km: float = 0.0,
20 | kind: str = "facility",
21 | ) -> str:
22 | """
23 | Create or update a ground location in the active scenario.
24 |
25 | Args:
26 | ctx: MCP request context (provides STK lifespan state).
27 | name: Object name (e.g., "Boulder").
28 | latitude_deg: Geodetic latitude in degrees [-90, 90].
29 | longitude_deg: Geodetic longitude in degrees [-180, 180].
30 | altitude_km: Altitude above mean sea level in kilometers.
31 | kind: Object kind: "facility" (default) or "place".
32 |
33 | Returns:
34 | Status message indicating success or error details.
35 |
36 | Examples:
37 | >>> create_location(ctx, name="Boulder", latitude_deg=40.015, longitude_deg=-105.27, altitude_km=1.656, kind="facility")
38 | "Successfully created facility: 'Boulder'"
39 | """
40 | lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
41 |
42 | try:
43 | scenario = lifespan_ctx.stk_root.CurrentScenario
44 | if scenario is None:
45 | return "Error: No active scenario found. Use 'setup_scenario' first."
46 | except Exception as e:
47 | return f"Error: Could not access current scenario: {e}"
48 |
49 | # Input validation
50 | if not (-90.0 <= latitude_deg <= 90.0):
51 | return "Error: latitude_deg must be within [-90, 90] degrees."
52 | if not (-180.0 <= longitude_deg <= 180.0):
53 | return "Error: longitude_deg must be within [-180, 180] degrees."
54 | if altitude_km < -0.5:
55 | return "Error: altitude_km must be >= -0.5 km."
56 | if kind.lower() not in ("facility", "place"):
57 | return "Error: kind must be 'facility' or 'place'."
58 |
59 | with STK_LOCK:
60 | ok, msg, _ = create_location_internal(
61 | stk_root=lifespan_ctx.stk_root,
62 | scenario=scenario,
63 | name=name,
64 | latitude_deg=latitude_deg,
65 | longitude_deg=longitude_deg,
66 | altitude_km=altitude_km,
67 | kind=kind,
68 | )
69 | return msg
70 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/location.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | import logging
4 | from typing import Literal
5 |
6 | from .core import stk_available, IAgStkObjectRoot, IAgScenario
7 | from .utils import timed_operation, safe_stk_command
8 |
9 | logger = logging.getLogger(__name__)
10 |
11 | try:
12 | # Enums are available on both Desktop and Engine
13 | from agi.stk12.stkobjects import AgESTKObjectType
14 | enums_available = True
15 | except Exception:
16 | AgESTKObjectType = None # type: ignore[assignment]
17 | enums_available = False
18 |
19 |
20 | @timed_operation
21 | def create_location_internal(
22 | stk_root: IAgStkObjectRoot,
23 | scenario: IAgScenario,
24 | name: str,
25 | latitude_deg: float,
26 | longitude_deg: float,
27 | altitude_km: float,
28 | kind: Literal["facility", "place"] = "facility",
29 | ):
30 | """
31 | Create or update a ground location in the active scenario.
32 |
33 | Uses the STK Object Model for creation and assigns position using object model
34 | where possible; falls back to STK Connect for position assignment to preserve
35 | cross-platform compatibility (Desktop/Engine).
36 |
37 | Returns:
38 | tuple[bool, str, object | None]: (success, message, created_or_found_object)
39 | """
40 | if not stk_available or not stk_root or not scenario:
41 | return False, "STK Root/Scenario is not available.", None
42 |
43 | kind = kind.lower().strip()
44 | if kind not in ("facility", "place"):
45 | return False, "Invalid kind. Use 'facility' or 'place'.", None
46 |
47 | if not enums_available:
48 | return False, "STK enums not available; cannot create object.", None
49 |
50 | obj_type = (
51 | AgESTKObjectType.eFacility if kind == "facility" else AgESTKObjectType.ePlace
52 | )
53 |
54 | try:
55 | children = scenario.Children
56 | if not children.Contains(obj_type, name):
57 | obj = children.New(obj_type, name)
58 | created = True
59 | else:
60 | obj = children.Item(name)
61 | created = False
62 |
63 | # Assign position via object model if available; otherwise use Connect
64 | try:
65 | # Most STK Python wrappers surface AssignGeodetic directly
66 | obj.Position.AssignGeodetic(latitude_deg, longitude_deg, altitude_km)
67 | except Exception:
68 | # Fallback to STK Connect with retry: Geodetic lat lon alt km
69 | class_name = "Facility" if kind == "facility" else "Place"
70 | cmd = (
71 | f"SetPosition */{class_name}/{name} Geodetic "
72 | f"{latitude_deg} {longitude_deg} {altitude_km} km"
73 | )
74 | safe_stk_command(stk_root, cmd)
75 |
76 | action = "created" if created else "updated"
77 | return True, f"Successfully {action} {kind}: '{name}'", obj
78 | except Exception as e:
79 | logger.error("Error creating %s '%s': %s", kind, name, e)
80 | return False, f"Error creating {kind} '{name}': {e}", None
81 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/analysis.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | import logging
4 | from typing import Any
5 |
6 | from .core import IAgStkObjectRoot
7 | from .utils import timed_operation
8 |
9 | logger = logging.getLogger(__name__)
10 |
11 |
12 | def _normalize_path(path: str) -> str:
13 | p = (path or "").strip()
14 | if not p:
15 | raise ValueError("Object path must be non-empty.")
16 | if p.startswith("*/"):
17 | return p
18 | if p.startswith("/"):
19 | return f"*{p}"
20 | # Require explicit class/name from users for reliability
21 | return f"*/{p}"
22 |
23 |
24 | @timed_operation
25 | def compute_access_intervals_internal(
26 | stk_root: IAgStkObjectRoot,
27 | object1_path: str,
28 | object2_path: str,
29 | ) -> dict[str, Any]:
30 | """Compute access intervals between two STK objects using the Object Model.
31 |
32 | Returns a dictionary with input paths and a list of {start, stop} intervals.
33 | """
34 | p1 = _normalize_path(object1_path)
35 | p2 = _normalize_path(object2_path)
36 |
37 | from_obj = stk_root.GetObjectFromPath(p1)
38 | to_obj = stk_root.GetObjectFromPath(p2)
39 |
40 | access = from_obj.GetAccessToObject(to_obj)
41 | access.ComputeAccess()
42 |
43 | intervals = access.AccessIntervals
44 | out: list[dict[str, str]] = []
45 | for i in range(intervals.Count):
46 | ivl = intervals.Item(i)
47 | out.append({"start": ivl.StartTime, "stop": ivl.StopTime})
48 |
49 | return {"from": p1, "to": p2, "intervals": out}
50 |
51 |
52 | @timed_operation
53 | def get_lla_ephemeris_internal(
54 | stk_root: IAgStkObjectRoot,
55 | satellite_path: str,
56 | step_sec: float = 60.0,
57 | ) -> dict[str, Any]:
58 | """Fetch LLA ephemeris for a satellite over the scenario interval using Data Providers.
59 |
60 | Returns a dictionary: {satellite, step_sec, records:[{time, lat_deg, lon_deg, alt_km}...]}
61 | """
62 | p = _normalize_path(satellite_path)
63 | sat = stk_root.GetObjectFromPath(p)
64 |
65 | scenario = stk_root.CurrentScenario
66 | if scenario is None:
67 | raise RuntimeError("No active scenario.")
68 |
69 | start = scenario.StartTime
70 | stop = scenario.StopTime
71 |
72 | # Data provider name and elements are standard for satellites
73 | dp_group = sat.DataProviders.Item("LLA State")
74 | dp = dp_group.Group.Item("Fixed")
75 | res = dp.ExecElements(start, stop, step_sec, ["Time", "Lat", "Lon", "Alt"])
76 |
77 | time_vals = list(res.DataSets.GetDataSetByName("Time").GetValues())
78 | lat_vals = list(res.DataSets.GetDataSetByName("Lat").GetValues())
79 | lon_vals = list(res.DataSets.GetDataSetByName("Lon").GetValues())
80 | alt_vals = list(res.DataSets.GetDataSetByName("Alt").GetValues())
81 |
82 | records: list[dict[str, float | str]] = []
83 | for i in range(len(time_vals)):
84 | records.append(
85 | {
86 | "time": time_vals[i],
87 | "lat_deg": float(lat_vals[i]),
88 | "lon_deg": float(lon_vals[i]),
89 | "alt_km": float(alt_vals[i]),
90 | }
91 | )
92 |
93 | return {"satellite": p, "step_sec": step_sec, "records": records}
94 |
95 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/satellite.py:
--------------------------------------------------------------------------------
```python
1 | import logging
2 | from mcp.server.fastmcp import Context
3 |
4 | # Use relative imports within the package
5 | from ..app import mcp_server # Import the server instance
6 | from ..stk_logic.core import StkState, STK_LOCK
7 | from ..stk_logic.decorators import require_stk_tool
8 | from ..stk_logic.satellite import create_satellite_internal
9 |
10 | logger = logging.getLogger(__name__)
11 |
12 | @mcp_server.tool() # Decorate with the server instance
13 | @require_stk_tool
14 | def create_satellite(
15 | ctx: Context,
16 | name: str,
17 | apogee_alt_km: float,
18 | perigee_alt_km: float,
19 | raan_deg: float,
20 | inclination_deg: float
21 | ) -> str:
22 | """
23 | MCP Tool: Creates/modifies an STK satellite using Apogee/Perigee altitudes, RAAN, and Inclination.
24 | Assumes a scenario is already open.
25 |
26 | Args:
27 | ctx: The MCP context.
28 | name: Desired name for the satellite.
29 | apogee_alt_km: Apogee altitude (km).
30 | perigee_alt_km: Perigee altitude (km).
31 | raan_deg: RAAN (degrees).
32 | inclination_deg: Inclination (degrees).
33 |
34 | Returns:
35 | A string indicating success or failure.
36 |
37 | Examples:
38 | >>> create_satellite(ctx, "ISS", apogee_alt_km=420, perigee_alt_km=410, raan_deg=51.6, inclination_deg=51.6)
39 | "Successfully created/configured satellite: 'ISS'"
40 | """
41 | logger.info("MCP Tool: create_satellite '%s'", name)
42 | lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
43 |
44 | # Get the current scenario from the STK root object
45 | try:
46 | scenario = lifespan_ctx.stk_root.CurrentScenario
47 | if scenario is None:
48 | return "Error: No active scenario found in STK. Use 'setup_scenario' tool first."
49 | logger.info(" Operating within scenario: %s", scenario.InstanceName)
50 | except Exception as e:
51 | return f"Error accessing current scenario: {e}. Use 'setup_scenario' tool first."
52 |
53 | # Input validation
54 | if apogee_alt_km < perigee_alt_km:
55 | return "Error: apogee_alt_km cannot be less than perigee_alt_km."
56 | if not (0.0 <= inclination_deg <= 180.0):
57 | return "Error: inclination_deg must be within [0, 180] degrees."
58 | # RAAN wraps; accept 0..360 inclusive
59 | if not (0.0 <= raan_deg <= 360.0):
60 | return "Error: raan_deg must be within [0, 360] degrees."
61 | if perigee_alt_km < -0.5 or apogee_alt_km < -0.5:
62 | return "Error: perigee/apogee altitudes must be >= -0.5 km."
63 |
64 | # Call the internal logic function
65 | try:
66 | with STK_LOCK:
67 | success, message, _ = create_satellite_internal(
68 | stk_root=lifespan_ctx.stk_root,
69 | scenario=scenario,
70 | name=name,
71 | apogee_alt_km=apogee_alt_km,
72 | perigee_alt_km=perigee_alt_km,
73 | raan_deg=raan_deg,
74 | inclination_deg=inclination_deg,
75 | )
76 | return message # Return the message from the internal function
77 |
78 | except ValueError as ve:
79 | error_msg = f"Configuration Error for satellite '{name}': {ve}"
80 | logger.error(" %s", error_msg)
81 | return error_msg
82 | except Exception as e:
83 | # Catch potential errors from the internal function (e.g., COM errors)
84 | error_msg = f"Error creating satellite '{name}': {e}"
85 | logger.error(" %s", error_msg)
86 | # import traceback
87 | # traceback.print_exc()
88 | return error_msg
89 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/cli.py:
--------------------------------------------------------------------------------
```python
1 | import os
2 | import sys
3 | import logging
4 | import uvicorn
5 | import typer
6 | from rich.console import Console
7 | from rich.table import Table
8 |
9 | # --- Check for STK installation early ---
10 | stk_installed = False
11 | try:
12 | from agi.stk12.stkengine import STKEngine # noqa: F401
13 | from agi.stk12.stkdesktop import STKDesktop # noqa: F401
14 | stk_installed = True
15 | except ImportError:
16 | print(
17 | "Warning: Ansys/AGI STK Python API not found. Please install it to use this application.",
18 | file=sys.stderr,
19 | )
20 | # Allow Typer to still show help, but commands will fail.
21 |
22 | # --- Local imports (safe regardless of STK availability) ---
23 | from stk_mcp.app import mcp_server
24 | from stk_mcp.stk_logic.core import create_stk_lifespan, StkMode # type: ignore
25 | from stk_mcp.stk_logic.config import get_config
26 | from stk_mcp.stk_logic.logging_config import configure_logging
27 | import anyio
28 |
29 |
30 | # --- Typer Application Setup ---
31 | app = typer.Typer(
32 | name="stk-mcp",
33 | help="A CLI for running and interacting with the STK-MCP server.",
34 | add_completion=False,
35 | )
36 | console = Console()
37 |
38 | def _validate_desktop_mode(mode: StkMode):
39 | """Callback to ensure 'desktop' mode is only used on Windows."""
40 | if mode == StkMode.DESKTOP and os.name != "nt":
41 | console.print(f"[bold red]Error:[/] STK Desktop mode is only available on Windows. Use '--mode {StkMode.ENGINE.value}'.")
42 | raise typer.Exit(code=1)
43 | return mode
44 |
45 | @app.command()
46 | def run(
47 | host: str = typer.Option(None, help="The host to bind the server to."),
48 | port: int = typer.Option(None, help="The port to run the server on."),
49 | mode: StkMode = typer.Option(
50 | StkMode.ENGINE if os.name != "nt" else StkMode.DESKTOP,
51 | "--mode", "-m",
52 | case_sensitive=False,
53 | help="STK execution mode. 'desktop' is only available on Windows.",
54 | callback=_validate_desktop_mode,
55 | ),
56 | log_level: str = typer.Option(
57 | "info",
58 | help="Log level: critical, error, warning, info, debug",
59 | ),
60 | ):
61 | """
62 | Run the STK-MCP server.
63 | """
64 | if not stk_installed:
65 | console.print("[bold red]Error:[/] Cannot run server. STK Python API is not installed.")
66 | raise typer.Exit(code=1)
67 |
68 | # Configure logging
69 | cfg = get_config()
70 | level = log_level or cfg.log_level
71 | configure_logging(level)
72 |
73 | # Resolve host/port from config if not provided
74 | host = host or cfg.default_host
75 | port = int(port or cfg.default_port)
76 |
77 | console.print(
78 | f"[green]Starting STK-MCP server in[/] [bold cyan]{mode.value}[/] [green]mode on {host}:{port}...[/]"
79 | )
80 |
81 | # Dynamically create the lifespan based on the selected mode
82 | stk_lifespan_manager = create_stk_lifespan(mode)
83 |
84 | # Attach the lifespan to the server instance
85 | mcp_server.lifespan = stk_lifespan_manager
86 |
87 | # Run the server using uvicorn
88 | uvicorn.run(
89 | mcp_server,
90 | host=host,
91 | port=port,
92 | )
93 |
94 | @app.command(name="list-tools")
95 | def list_tools():
96 | """
97 | List all available MCP tools and their descriptions.
98 | """
99 | if not stk_installed:
100 | console.print("[bold red]Error:[/] Cannot list tools. STK Python API is not installed.")
101 | raise typer.Exit(code=1)
102 |
103 | table = Table(title="[bold blue]STK-MCP Available Tools[/bold blue]")
104 | table.add_column("Tool Name", style="cyan", no_wrap=True)
105 | table.add_column("Description", style="magenta")
106 |
107 | tools = anyio.run(mcp_server.list_tools)
108 | if not tools:
109 | console.print("[yellow]No tools have been registered on the server.[/yellow]")
110 | return
111 |
112 | for t in tools:
113 | description = (t.description or "No description provided.").strip()
114 | table.add_row(t.name, description)
115 |
116 | console.print(table)
117 |
118 |
119 | if __name__ == "__main__":
120 | app()
121 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/objects.py:
--------------------------------------------------------------------------------
```python
1 | from __future__ import annotations
2 |
3 | import logging
4 | from typing import Optional
5 |
6 | from .core import stk_available, IAgStkObjectRoot
7 | from .utils import safe_exec_lines, timed_operation
8 |
9 | logger = logging.getLogger(__name__)
10 |
11 |
12 | def _exec_lines(stk_root: IAgStkObjectRoot, cmd: str) -> list[str]:
13 | """
14 | Execute an STK Connect command and return the result lines.
15 | Safe helper that returns an empty list on failure.
16 | """
17 | return safe_exec_lines(stk_root, cmd)
18 |
19 |
20 | def _parse_all_instance_names(lines: list[str]) -> list[tuple[str, str]]:
21 | """
22 | Parse lines returned by the "AllInstanceNames" Connect command.
23 |
24 | Returns a list of tuples: (class_name, instance_name)
25 | """
26 | out: list[tuple[str, str]] = []
27 | for raw in lines:
28 | line = (raw or "").strip()
29 | if not line:
30 | continue
31 | # Heuristically skip headers if present
32 | lower = line.lower()
33 | if "number" in lower and ("object" in lower or "instance" in lower):
34 | continue
35 |
36 | # Paths often look like "/Class/Name" or "*/Class/Name" or longer
37 | parts = [p for p in line.split('/') if p]
38 | if len(parts) >= 2:
39 | cls = parts[-2]
40 | name = parts[-1]
41 | out.append((cls, name))
42 | else:
43 | # Fallback if the format is unexpected
44 | out.append(("", line))
45 | return out
46 |
47 |
48 | _TOP_LEVEL_CLASSES = (
49 | # Common top-level object classes in STK
50 | "Satellite",
51 | "Facility",
52 | "Place",
53 | "Aircraft",
54 | "Ship",
55 | "GroundVehicle",
56 | "Missile",
57 | "LaunchVehicle",
58 | "Submarine",
59 | "AreaTarget",
60 | "LineTarget",
61 | )
62 |
63 | _SENSOR_PARENTS = (
64 | # Object classes that commonly host Sensors
65 | "Satellite",
66 | "Facility",
67 | "Aircraft",
68 | "Ship",
69 | "GroundVehicle",
70 | "Missile",
71 | "LaunchVehicle",
72 | "Submarine",
73 | )
74 |
75 |
76 | def _normalize_filter(filter_type: str) -> set[str] | None:
77 | """
78 | Normalize a user-provided filter type (case-insensitive) to a set of
79 | canonical STK class names. Returns None if the filter is unrecognized.
80 | """
81 | t = (filter_type or "").strip().lower()
82 | if not t:
83 | return None
84 |
85 | # Aliases and plural handling
86 | mapping: dict[str, set[str]] = {
87 | "sat": {"Satellite"},
88 | "satellite": {"Satellite"},
89 | "satellites": {"Satellite"},
90 | "facility": {"Facility"},
91 | "facilities": {"Facility"},
92 | "place": {"Place"},
93 | "places": {"Place"},
94 | "location": {"Facility", "Place"},
95 | "locations": {"Facility", "Place"},
96 | "sensor": {"Sensor"},
97 | "sensors": {"Sensor"},
98 | "aircraft": {"Aircraft"},
99 | "ship": {"Ship"},
100 | "ships": {"Ship"},
101 | "groundvehicle": {"GroundVehicle"},
102 | "groundvehicles": {"GroundVehicle"},
103 | "missile": {"Missile"},
104 | "missiles": {"Missile"},
105 | "areatarget": {"AreaTarget"},
106 | "areatargets": {"AreaTarget"},
107 | "linetarget": {"LineTarget"},
108 | "linetargets": {"LineTarget"},
109 | "launchvehicle": {"LaunchVehicle"},
110 | "launchvehicles": {"LaunchVehicle"},
111 | "submarine": {"Submarine"},
112 | "submarines": {"Submarine"},
113 | }
114 | return mapping.get(t)
115 |
116 |
117 | @timed_operation
118 | def list_objects_internal(
119 | stk_root: IAgStkObjectRoot,
120 | filter_type: Optional[str] = None,
121 | ) -> list[dict[str, str]]:
122 | """
123 | Enumerate objects in the active scenario and return a list of
124 | {"name": <instance name>, "type": <class>} dictionaries.
125 |
126 | Uses STK Connect (AllInstanceNames) for broad compatibility with
127 | both STK Desktop and STK Engine.
128 | """
129 | if not stk_available or not stk_root:
130 | raise RuntimeError("STK Root is not available.")
131 |
132 | # Ensure there is an active scenario
133 | try:
134 | scenario = stk_root.CurrentScenario
135 | if scenario is None:
136 | raise RuntimeError("No active scenario found.")
137 | except Exception as e:
138 | raise RuntimeError(f"Could not access current scenario: {e}")
139 |
140 | normalized: set[str] | None = _normalize_filter(filter_type) if filter_type else None
141 |
142 | results: list[dict[str, str]] = []
143 |
144 | def add_from_cmd(cmd: str, expected_type: str | None = None) -> None:
145 | lines = _exec_lines(stk_root, cmd)
146 | for cls, name in _parse_all_instance_names(lines):
147 | typ = expected_type or (cls if cls else "")
148 | if not typ:
149 | continue
150 | if normalized is not None and typ not in normalized:
151 | continue
152 | results.append({"name": name, "type": typ})
153 |
154 | # Top-level classes
155 | for cls in _TOP_LEVEL_CLASSES:
156 | if normalized is not None and cls not in normalized:
157 | continue
158 | add_from_cmd(f"AllInstanceNames */{cls}", expected_type=cls)
159 |
160 | # Sensors (nested under multiple parents)
161 | # Only include if no filter, or if filter explicitly asks for sensors
162 | if normalized is None or ("Sensor" in normalized):
163 | for parent in _SENSOR_PARENTS:
164 | add_from_cmd(f"AllInstanceNames */{parent}/*/Sensor", expected_type="Sensor")
165 |
166 | return results
167 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/satellite.py:
--------------------------------------------------------------------------------
```python
1 | import os
2 | import logging
3 | from . import core as core
4 | from .core import IAgStkObjectRoot, IAgScenario
5 | from .utils import timed_operation
6 | from .config import get_config
7 |
8 | logger = logging.getLogger(__name__)
9 |
10 | # Import STK Objects specific to satellite creation if available (Windows Desktop/Engine)
11 | AgESTKObjectType = None
12 | AgEVePropagatorType = None
13 | AgEClassicalLocation = None
14 | win32com_client = None
15 | satellite_capable = False
16 |
17 | if core.stk_available and os.name == 'nt':
18 | try:
19 | from agi.stk12.stkobjects import (
20 | AgESTKObjectType as AgESTKObjectTypeImport,
21 | AgEVePropagatorType as AgEVePropagatorTypeImport,
22 | AgEClassicalLocation as AgEClassicalLocationImport,
23 | )
24 | import win32com.client as win32com_client_import
25 | AgESTKObjectType = AgESTKObjectTypeImport
26 | AgEVePropagatorType = AgEVePropagatorTypeImport
27 | AgEClassicalLocation = AgEClassicalLocationImport
28 | win32com_client = win32com_client_import
29 | satellite_capable = True
30 | except ImportError:
31 | logger.warning("Could not import STK COM types for satellite creation (Windows only feature).")
32 | except Exception as e:
33 | logger.error("Error importing win32com or STK enums: %s", e)
34 |
35 |
36 | cfg = get_config()
37 | EARTH_RADIUS_KM = cfg.earth_radius_km
38 |
39 | @timed_operation
40 | def create_satellite_internal(
41 | stk_root: IAgStkObjectRoot, # Although not directly used, good for context
42 | scenario: IAgScenario,
43 | name: str,
44 | apogee_alt_km: float,
45 | perigee_alt_km: float,
46 | raan_deg: float,
47 | inclination_deg: float,
48 | ):
49 | """
50 | Internal logic to create/configure an STK satellite.
51 |
52 | Returns:
53 | tuple: (success_flag, status_message, satellite_object_or_None)
54 |
55 | Raises:
56 | ValueError: If input parameters are invalid (e.g., apogee < perigee).
57 | Exception: For COM or other STK errors.
58 | """
59 | if not core.stk_available or not scenario or win32com_client is None or not satellite_capable:
60 | raise RuntimeError("STK modules, active scenario, or win32com not available/initialized.")
61 | if AgESTKObjectType is None or AgEVePropagatorType is None or AgEClassicalLocation is None:
62 | raise RuntimeError("Required STK Object Enums not imported.")
63 |
64 | logger.info(" Attempting internal satellite creation/configuration: %s", name)
65 |
66 | if apogee_alt_km < perigee_alt_km:
67 | raise ValueError("Apogee altitude cannot be less than Perigee altitude.")
68 |
69 | # --- Calculate Semi-Major Axis (a) and Eccentricity (e) ---
70 | radius_apogee_km = apogee_alt_km + EARTH_RADIUS_KM
71 | radius_perigee_km = perigee_alt_km + EARTH_RADIUS_KM
72 | semi_major_axis_km = (radius_apogee_km + radius_perigee_km) / 2.0
73 | denominator = radius_apogee_km + radius_perigee_km
74 | eccentricity = 0.0 if denominator == 0 else (radius_apogee_km - radius_perigee_km) / denominator
75 |
76 | logger.debug(" Calculated Semi-Major Axis (a): %.3f km", semi_major_axis_km)
77 | logger.debug(" Calculated Eccentricity (e): %.6f", eccentricity)
78 |
79 | # --- Get or Create Satellite Object ---
80 | scenario_children = scenario.Children
81 | satellite = None
82 | if not scenario_children.Contains(AgESTKObjectType.eSatellite, name):
83 | logger.info(" Creating new Satellite object: %s", name)
84 | satellite = scenario_children.New(AgESTKObjectType.eSatellite, name)
85 | else:
86 | logger.info(" Satellite '%s' already exists. Getting reference.", name)
87 | satellite = scenario_children.Item(name)
88 |
89 | if satellite is None:
90 | raise Exception(f"Failed to create or retrieve satellite object '{name}'.")
91 |
92 | # --- Set Propagator to TwoBody ---
93 | logger.info(" Setting propagator to TwoBody...")
94 | satellite.SetPropagatorType(AgEVePropagatorType.ePropagatorTwoBody)
95 | propagator = satellite.Propagator
96 |
97 | propagator_twobody = win32com_client.CastTo(propagator, "IAgVePropagatorTwoBody")
98 | if propagator_twobody is None:
99 | raise Exception("Failed to cast propagator to IAgVePropagatorTwoBody.")
100 |
101 | # --- Define Orbital Elements ---
102 | argp_deg = 0.0 # Assumed
103 | true_anom_deg = 0.0 # Assumed (starts at perigee)
104 |
105 | logger.info(" Assigning Classical Elements (J2000):")
106 | # (Print statements omitted for brevity, add back if desired)
107 |
108 | orbit_state = propagator_twobody.InitialState.Representation
109 | classical_elements = win32com_client.CastTo(orbit_state, "IAgOrbitStateClassical")
110 |
111 | if classical_elements:
112 | classical_elements.AssignClassical(
113 | AgEClassicalLocation.eCoordinateSystemJ2000,
114 | semi_major_axis_km, eccentricity, inclination_deg,
115 | argp_deg, raan_deg, true_anom_deg
116 | )
117 | else:
118 | raise Exception("Failed to cast orbit state to IAgOrbitStateClassical.")
119 |
120 | # --- Propagate the Orbit ---
121 | logger.info(" Propagating orbit...")
122 | propagator_twobody.Propagate()
123 |
124 | logger.info(" Internal satellite configuration for '%s' complete.", name)
125 | # Return success flag, message, and the object
126 | return True, f"Successfully created/configured satellite: '{satellite.InstanceName}'", satellite
127 |
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/core.py:
--------------------------------------------------------------------------------
```python
1 | import os
2 | import platform
3 | import logging
4 | from contextlib import asynccontextmanager
5 | from collections.abc import AsyncIterator
6 | from typing import Union
7 | from enum import Enum # <--- IMPORT Enum
8 | from threading import Lock
9 |
10 | from pydantic import BaseModel
11 | from typing import NamedTuple, Optional
12 | from mcp.server.fastmcp import FastMCP
13 |
14 | logger = logging.getLogger(__name__)
15 |
16 | # --- Define shared data types here ---
17 |
18 | class StkMode(str, Enum):
19 | """Enumeration for selecting the STK execution mode."""
20 | DESKTOP = "desktop"
21 | ENGINE = "engine"
22 |
23 | # --- Attempt STK Imports ---
24 | stk_available = False
25 | STKApplication = None
26 | STKDesktop = None
27 | STKEngine = None
28 | IAgStkObjectRoot = None
29 | IAgScenario = None
30 |
31 | _platform = platform.system()
32 | if _platform == "Windows":
33 | try:
34 | from agi.stk12.stkdesktop import STKDesktop as STKDesktopImport, STKApplication as STKApplicationImport
35 | from agi.stk12.stkengine import STKEngine as STKEngineImport
36 | from agi.stk12.stkobjects import IAgStkObjectRoot as IAgStkObjectRootImport, IAgScenario as IAgScenarioImport
37 |
38 | STKDesktop = STKDesktopImport
39 | STKApplication = STKApplicationImport # Desktop App
40 | STKEngine = STKEngineImport
41 | IAgStkObjectRoot = IAgStkObjectRootImport
42 | IAgScenario = IAgScenarioImport
43 | stk_available = True
44 | logger.info("STK Desktop and Engine modules loaded successfully (Windows).")
45 | except ImportError:
46 | logger.warning("Failed to import STK modules on Windows. Functionality disabled.")
47 | elif _platform == "Linux":
48 | try:
49 | from agi.stk12.stkengine import STKEngine as STKEngineImport, IAgStkObjectRoot as IAgStkObjectRootImport
50 | from agi.stk12.stkobjects import IAgScenario as IAgScenarioImport
51 |
52 | STKEngine = STKEngineImport
53 | IAgStkObjectRoot = IAgStkObjectRootImport
54 | IAgScenario = IAgScenarioImport
55 | stk_available = True
56 | logger.info("STK Engine modules loaded successfully (Linux). STK Desktop is unavailable.")
57 | except ImportError:
58 | logger.warning("Failed to import STK Engine modules. Functionality disabled.")
59 | elif _platform == "Darwin":
60 | # STK Engine is not supported on macOS
61 | logger.error("Detected macOS (Darwin). STK Engine/Desktop are not supported on this platform.")
62 | else:
63 | logger.warning("Unknown platform '%s'. STK availability undetermined.", _platform)
64 |
65 | # A type hint for whichever application object is in use
66 | StkAppType = object | None
67 |
68 | class StkState(BaseModel):
69 | """Holds the state of the STK application connection."""
70 | stk_app: StkAppType = None
71 | stk_root: object | None = None
72 | mode: StkMode | None = None
73 |
74 | # Global lock to serialize all STK access across tools/resources
75 | STK_LOCK: Lock = Lock()
76 |
77 |
78 | class OperationResult(NamedTuple):
79 | """Standard result type for STK operations."""
80 | success: bool
81 | message: str
82 | data: Optional[dict] = None
83 |
84 | def create_stk_lifespan(mode: StkMode):
85 | """
86 | A factory that returns an async context manager for the STK lifecycle.
87 | """
88 | @asynccontextmanager
89 | async def stk_lifespan_manager(server: FastMCP) -> AsyncIterator[StkState]:
90 | """
91 | Manages the STK application lifecycle based on the selected mode.
92 | """
93 | if not stk_available or IAgStkObjectRoot is None:
94 | logger.warning("STK is not available. MCP server will run without STK functionality.")
95 | yield StkState(mode=mode)
96 | return
97 |
98 | logger.info("MCP Server Startup: Initializing STK in '%s' mode...", mode.value)
99 | state = StkState(mode=mode)
100 |
101 | try:
102 | if mode == StkMode.DESKTOP:
103 | # --- Desktop Mode Logic ---
104 | logger.info(" Attempting to attach to existing STK instance...")
105 | try:
106 | state.stk_app = STKDesktop.AttachToApplication()
107 | logger.info(" Successfully attached to existing STK instance.")
108 | state.stk_app.Visible = True
109 | except Exception:
110 | logger.info(" Could not attach. Launching new STK instance...")
111 | state.stk_app = STKDesktop.StartApplication(visible=True, userControl=True)
112 |
113 | state.stk_root = state.stk_app.Root
114 | # Close any open scenario to start clean
115 | if state.stk_root and state.stk_root.Children.Count > 0:
116 | logger.info(" Closing existing scenario '%s'...", state.stk_root.CurrentScenario.InstanceName)
117 | state.stk_root.CloseScenario()
118 |
119 | elif mode == StkMode.ENGINE:
120 | # --- Engine Mode Logic ---
121 | logger.info(" Starting new STK Engine instance...")
122 | state.stk_app = STKEngine.StartApplication(noGraphics=True)
123 | state.stk_root = state.stk_app.NewObjectRoot()
124 | logger.info(" STK Engine instance started.")
125 |
126 | if state.stk_root is None:
127 | raise RuntimeError("Failed to obtain STK Root object.")
128 |
129 | logger.info("STK Initialized. Providing STK context to tools.")
130 | yield state
131 |
132 | except Exception as e:
133 | logger.exception("FATAL: Failed to initialize STK in %s mode: %s", mode.value, e)
134 | yield StkState(mode=mode) # Yield empty state on failure
135 |
136 | finally:
137 | logger.info("MCP Server Shutdown: Cleaning up STK (%s mode)...", mode.value)
138 | if state.stk_app:
139 | try:
140 | state.stk_app.Close()
141 | logger.info(" STK Application/Engine Closed.")
142 | except Exception as quit_e:
143 | logger.warning(" Error closing STK: %s", quit_e)
144 | logger.info("STK Cleanup Complete.")
145 |
146 | return stk_lifespan_manager
147 |
```